Skip to content

Cdf processor

Defines change data feed processor behaviour.

CDFProcessor

Change data feed processor class.

Source code in mkdocs/lakehouse_engine/packages/terminators/cdf_processor.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class CDFProcessor:
    """Change data feed processor class."""

    _logger = LoggingHandler(__name__).get_logger()

    @classmethod
    def expose_cdf(cls, spec: TerminatorSpec) -> None:
        """Expose CDF to external location.

        Args:
            spec: terminator specification.
        """
        args = spec.args or {}
        cls._logger.info("Reading CDF from input table...")

        df_cdf = ReaderFactory.get_data(cls._get_table_cdf_input_specs(spec))
        new_df_cdf = df_cdf.withColumn(
            "_commit_timestamp",
            date_format(col("_commit_timestamp"), "yyyyMMddHHmmss"),
        )

        cls._logger.info("Writing CDF to external table...")
        cls._write_cdf_to_external(
            spec,
            new_df_cdf.repartition(
                args.get("materialized_cdf_num_partitions", col("_commit_timestamp"))
            ),
        )

        # used to delete old data on CDF table (don't remove parquet).
        if args.get("clean_cdf", True):
            cls._logger.info("Cleaning CDF table...")
            cls.delete_old_data(spec)

        # used to delete old parquet files.
        if args.get("vacuum_cdf", False):
            cls._logger.info("Vacuuming CDF table...")
            cls.vacuum_cdf_data(spec)

    @staticmethod
    def _write_cdf_to_external(
        spec: TerminatorSpec, df: DataFrame, data: OrderedDict | None = None
    ) -> None:
        """Write cdf results dataframe.

        Args:
            spec: terminator specification.
            df: dataframe with cdf results to write.
            data: list of all dfs generated on previous steps before writer.
        """
        args = spec.args or {}
        materialized_cdf_location = args.get("materialized_cdf_location")
        if not materialized_cdf_location:
            raise ValueError(ERROR_MESSAGE)
        WriterFactory.get_writer(
            spec=OutputSpec(
                spec_id="materialized_cdf",
                input_id="input_table",
                location=materialized_cdf_location,
                write_type=WriteType.APPEND.value,
                data_format=args.get("data_format", OutputFormat.DELTAFILES.value),
                options=args.get("materialized_cdf_options"),
                partitions=["_commit_timestamp"],
            ),
            df=df,
            data=data,
        ).write()

    @staticmethod
    def _get_table_cdf_input_specs(spec: TerminatorSpec) -> InputSpec:
        """Get the input specifications from a terminator spec.

        Args:
            spec: terminator specifications.

        Returns:
            List of input specifications.
        """
        args = spec.args or {}
        options = {
            "readChangeFeed": "true",
            **args.get("db_table_options", {}),
        }

        db_table = args.get("db_table")
        if not db_table:
            raise ValueError("db_table must be provided.")

        input_specs = InputSpec(
            spec_id="input_table",
            db_table=db_table,
            read_type=ReadType.STREAMING.value,
            data_format=OutputFormat.DELTAFILES.value,
            options=options,
        )

        return input_specs

    @classmethod
    def delete_old_data(cls, spec: TerminatorSpec) -> None:
        """Delete old data from cdf delta table.

        Args:
            spec: terminator specifications.
        """
        args = spec.args or {}
        today_datetime = datetime.today()
        limit_date = today_datetime + timedelta(days=args.get("days_to_keep", 30) * -1)
        limit_timestamp = limit_date.strftime("%Y%m%d%H%M%S")

        materialized_cdf_location = args.get("materialized_cdf_location")
        if not materialized_cdf_location:
            raise ValueError(ERROR_MESSAGE)

        cdf_delta_table = DeltaTable.forPath(ExecEnv.SESSION, materialized_cdf_location)

        cdf_delta_table.delete(col("_commit_timestamp") < limit_timestamp)

    @classmethod
    def vacuum_cdf_data(cls, spec: TerminatorSpec) -> None:
        """Vacuum old data from cdf delta table.

        Args:
            spec: terminator specifications.
        """
        args = spec.args or {}
        materialized_cdf_location = args.get("materialized_cdf_location")
        if not materialized_cdf_location:
            raise ValueError(ERROR_MESSAGE)
        cdf_delta_table = DeltaTable.forPath(ExecEnv.SESSION, materialized_cdf_location)

        cdf_delta_table.vacuum(args.get("vacuum_hours", 168))

delete_old_data(spec) classmethod

Delete old data from cdf delta table.

Parameters:

Name Type Description Default
spec TerminatorSpec

terminator specifications.

 required
Source code in mkdocs/lakehouse_engine/packages/terminators/cdf_processor.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@classmethod
def delete_old_data(cls, spec: TerminatorSpec) -> None:
    """Delete old data from cdf delta table.

    Args:
        spec: terminator specifications.
    """
    args = spec.args or {}
    today_datetime = datetime.today()
    limit_date = today_datetime + timedelta(days=args.get("days_to_keep", 30) * -1)
    limit_timestamp = limit_date.strftime("%Y%m%d%H%M%S")

    materialized_cdf_location = args.get("materialized_cdf_location")
    if not materialized_cdf_location:
        raise ValueError(ERROR_MESSAGE)

    cdf_delta_table = DeltaTable.forPath(ExecEnv.SESSION, materialized_cdf_location)

    cdf_delta_table.delete(col("_commit_timestamp") < limit_timestamp)

expose_cdf(spec) classmethod

Expose CDF to external location.

Parameters:

Name Type Description Default
spec TerminatorSpec

terminator specification.

 required
Source code in mkdocs/lakehouse_engine/packages/terminators/cdf_processor.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@classmethod
def expose_cdf(cls, spec: TerminatorSpec) -> None:
    """Expose CDF to external location.

    Args:
        spec: terminator specification.
    """
    args = spec.args or {}
    cls._logger.info("Reading CDF from input table...")

    df_cdf = ReaderFactory.get_data(cls._get_table_cdf_input_specs(spec))
    new_df_cdf = df_cdf.withColumn(
        "_commit_timestamp",
        date_format(col("_commit_timestamp"), "yyyyMMddHHmmss"),
    )

    cls._logger.info("Writing CDF to external table...")
    cls._write_cdf_to_external(
        spec,
        new_df_cdf.repartition(
            args.get("materialized_cdf_num_partitions", col("_commit_timestamp"))
        ),
    )

    # used to delete old data on CDF table (don't remove parquet).
    if args.get("clean_cdf", True):
        cls._logger.info("Cleaning CDF table...")
        cls.delete_old_data(spec)

    # used to delete old parquet files.
    if args.get("vacuum_cdf", False):
        cls._logger.info("Vacuuming CDF table...")
        cls.vacuum_cdf_data(spec)

vacuum_cdf_data(spec) classmethod

Vacuum old data from cdf delta table.

Parameters:

Name Type Description Default
spec TerminatorSpec

terminator specifications.

 required
Source code in mkdocs/lakehouse_engine/packages/terminators/cdf_processor.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@classmethod
def vacuum_cdf_data(cls, spec: TerminatorSpec) -> None:
    """Vacuum old data from cdf delta table.

    Args:
        spec: terminator specifications.
    """
    args = spec.args or {}
    materialized_cdf_location = args.get("materialized_cdf_location")
    if not materialized_cdf_location:
        raise ValueError(ERROR_MESSAGE)
    cdf_delta_table = DeltaTable.forPath(ExecEnv.SESSION, materialized_cdf_location)

    cdf_delta_table.vacuum(args.get("vacuum_hours", 168))