Write to Delta

SparkPCDMToDeltaDestination

Bases: DestinationInterface

The Process Control Data Model written to Delta

Parameters:

Name	Type	Description	Default
data	DataFrame	Dataframe to be merged into a Delta Table	required
options	dict	Options that can be specified for a Delta Table read operation (See Attributes table below). Further information on the options is available for batch and streaming.	required
destination_float	str	Either the name of the Hive Metastore or Unity Catalog Delta Table or the path to the Delta table to store float values.	required
destination_string	Optional str	Either the name of the Hive Metastore or Unity Catalog Delta Table or the path to the Delta table to store string values.	None
destination_integer	Optional str	Either the name of the Hive Metastore or Unity Catalog Delta Table or the path to the Delta table to store integer values	None
mode	str	Method of writing to Delta Table - append/overwrite (batch), append/complete (stream)	None
trigger	str	Frequency of the write operation. Specify "availableNow" to execute a trigger once, otherwise specify a time period such as "30 seconds", "5 minutes"	'10 seconds'
query_name	str	Unique name for the query in associated SparkSession	'PCDMToDeltaDestination'
merge	bool	Use Delta Merge to perform inserts, updates and deletes	True
try_broadcast_join	bool	Attempts to perform a broadcast join in the merge which can leverage data skipping using partition pruning and file pruning automatically. Can fail if dataframe being merged is large and therefore more suitable for streaming merges than batch merges	False
remove_nanoseconds	bool	Removes nanoseconds from the EventTime column and replaces with zeros	False
remove_duplicates	bool	Removes duplicates before writing the data	True

Attributes:

Name	Type	Description
checkpointLocation	str	Path to checkpoint files. (Streaming)

Source code in src/sdk/python/rtdip_sdk/pipelines/destinations/spark/pcdm_to_delta.py

class SparkPCDMToDeltaDestination(DestinationInterface):
    """
    The Process Control Data Model written to Delta

    Args:
        data (DataFrame): Dataframe to be merged into a Delta Table
        options (dict): Options that can be specified for a Delta Table read operation (See Attributes table below). Further information on the options is available for [batch](https://docs.delta.io/latest/delta-batch.html#write-to-a-table){ target="_blank" } and [streaming](https://docs.delta.io/latest/delta-streaming.html#delta-table-as-a-sink){ target="_blank" }.
        destination_float (str): Either the name of the Hive Metastore or Unity Catalog Delta Table **or** the path to the Delta table to store float values.
        destination_string (Optional str): Either the name of the Hive Metastore or Unity Catalog Delta Table **or** the path to the Delta table to store string values.
        destination_integer (Optional str): Either the name of the Hive Metastore or Unity Catalog Delta Table **or** the path to the Delta table to store integer values
        mode (str): Method of writing to Delta Table - append/overwrite (batch), append/complete (stream)
        trigger (str): Frequency of the write operation. Specify "availableNow" to execute a trigger once, otherwise specify a time period such as "30 seconds", "5 minutes"
        query_name (str): Unique name for the query in associated SparkSession
        merge (bool): Use Delta Merge to perform inserts, updates and deletes
        try_broadcast_join (bool): Attempts to perform a broadcast join in the merge which can leverage data skipping using partition pruning and file pruning automatically. Can fail if dataframe being merged is large and therefore more suitable for streaming merges than batch merges
        remove_nanoseconds (bool): Removes nanoseconds from the EventTime column and replaces with zeros
        remove_duplicates (bool: Removes duplicates before writing the data

    Attributes:
        checkpointLocation (str): Path to checkpoint files. (Streaming)
    """

    spark: SparkSession
    data: DataFrame
    options: dict
    destination_float: str
    destination_string: str
    destination_integer: str
    mode: str
    trigger: str
    query_name: str
    merge: bool
    try_broadcast_join: bool
    remove_nanoseconds: bool
    remove_duplicates: bool

    def __init__(
        self,
        spark: SparkSession,
        data: DataFrame,
        options: dict,
        destination_float: str,
        destination_string: str = None,
        destination_integer: str = None,
        mode: str = None,
        trigger="10 seconds",
        query_name: str = "PCDMToDeltaDestination",
        merge: bool = True,
        try_broadcast_join=False,
        remove_nanoseconds: bool = False,
        remove_duplicates: bool = True,
    ) -> None:
        self.spark = spark
        self.data = data
        self.destination_float = destination_float
        self.destination_string = destination_string
        self.destination_integer = destination_integer
        self.options = options
        self.mode = mode
        self.trigger = trigger
        self.query_name = query_name
        self.merge = merge
        self.try_broadcast_join = try_broadcast_join
        self.remove_nanoseconds = remove_nanoseconds
        self.remove_duplicates = remove_duplicates

    @staticmethod
    def system_type():
        """
        Attributes:
            SystemType (Environment): Requires PYSPARK
        """
        return SystemType.PYSPARK

    @staticmethod
    def libraries():
        libraries = Libraries()
        libraries.add_maven_library(get_default_package("spark_delta_core"))
        return libraries

    @staticmethod
    def settings() -> dict:
        return {}

    def pre_write_validation(self):
        return True

    def post_write_validation(self):
        return True

    def _get_eventdate_string(self, df: DataFrame) -> str:
        dates_df = df.select("EventDate").distinct()
        dates_df = dates_df.select(
            date_format("EventDate", "yyyy-MM-dd").alias("EventDate")
        )
        dates_list = list(dates_df.toPandas()["EventDate"])
        return str(dates_list).replace("[", "").replace("]", "")

    def _write_delta_merge(self, df: DataFrame, destination: str):
        df = df.select(
            "EventDate", "TagName", "EventTime", "Status", "Value", "ChangeType"
        )
        when_matched_update_list = [
            DeltaMergeConditionValues(
                condition="(source.ChangeType IN ('insert', 'update', 'upsert')) AND ((source.Status != target.Status) OR (source.Value != target.Value))",
                values={
                    "EventDate": "source.EventDate",
                    "TagName": "source.TagName",
                    "EventTime": "source.EventTime",
                    "Status": "source.Status",
                    "Value": "source.Value",
                },
            )
        ]
        when_matched_delete_list = [
            DeltaMergeCondition(condition="source.ChangeType = 'delete'")
        ]
        when_not_matched_insert_list = [
            DeltaMergeConditionValues(
                condition="(source.ChangeType IN ('insert', 'update', 'upsert'))",
                values={
                    "EventDate": "source.EventDate",
                    "TagName": "source.TagName",
                    "EventTime": "source.EventTime",
                    "Status": "source.Status",
                    "Value": "source.Value",
                },
            )
        ]

        merge_condition = "source.EventDate = target.EventDate AND source.TagName = target.TagName AND source.EventTime = target.EventTime"

        perform_merge = True
        if self.try_broadcast_join != True:
            eventdate_string = self._get_eventdate_string(df)
            if eventdate_string == None or eventdate_string == "":
                perform_merge = False
            else:
                merge_condition = (
                    "target.EventDate in ({}) AND ".format(eventdate_string)
                    + merge_condition
                )

        if perform_merge == True:
            SparkDeltaMergeDestination(
                spark=self.spark,
                data=df,
                destination=destination,
                options=self.options,
                merge_condition=merge_condition,
                when_matched_update_list=when_matched_update_list,
                when_matched_delete_list=when_matched_delete_list,
                when_not_matched_insert_list=when_not_matched_insert_list,
                try_broadcast_join=self.try_broadcast_join,
                trigger=self.trigger,
                query_name=self.query_name,
            ).write_batch()

    def _write_delta_batch(self, df: DataFrame, destination: str):
        if self.merge == True:
            if "EventDate" not in df.columns:
                df = df.withColumn("EventDate", date_format("EventTime", "yyyy-MM-dd"))

            self._write_delta_merge(
                df.filter(col("ChangeType").isin("insert", "update", "upsert")),
                destination,
            )
            self._write_delta_merge(
                df.filter(col("ChangeType") == "delete"), destination
            )
        else:
            df = df.select("TagName", "EventTime", "Status", "Value")
            SparkDeltaDestination(
                data=df,
                destination=destination,
                options=self.options,
                mode=self.mode,
                trigger=self.trigger,
                query_name=self.query_name,
            ).write_batch()

    def _write_data_by_type(self, df: DataFrame):
        if self.merge == True:
            df = df.withColumn(
                "ChangeType",
                when(df["ChangeType"].isin("insert", "update"), "upsert").otherwise(
                    df["ChangeType"]
                ),
            )

        if self.remove_nanoseconds == True:
            df = df.withColumn(
                "EventTime",
                (floor(col("EventTime").cast("double") * 1000) / 1000).cast(
                    "timestamp"
                ),
            )

        if self.remove_duplicates == True:
            df = df.drop_duplicates(["TagName", "EventTime", "ChangeType"])

        float_df = df.filter(ValueTypeConstants.FLOAT_VALUE).withColumn(
            "Value", col("Value").cast("float")
        )
        self._write_delta_batch(float_df, self.destination_float)

        if self.destination_string != None:
            string_df = df.filter(ValueTypeConstants.STRING_VALUE)
            self._write_delta_batch(string_df, self.destination_string)

        if self.destination_integer != None:
            integer_df = df.filter(ValueTypeConstants.INTEGER_VALUE).withColumn(
                "Value", col("Value").cast("integer")
            )
            self._write_delta_batch(integer_df, self.destination_integer)

    def _write_stream_microbatches(self, df: DataFrame, epoch_id=None):  # NOSONAR
        df.persist()
        self._write_data_by_type(df)
        df.unpersist()

    def write_batch(self):
        """
        Writes Process Control Data Model data to Delta
        """
        try:
            if self.try_broadcast_join != True:
                self.data.persist()

            self._write_data_by_type(self.data)

            if self.try_broadcast_join != True:
                self.data.unpersist()

        except Py4JJavaError as e:
            logging.exception(e.errmsg)
            raise e
        except Exception as e:
            logging.exception(str(e))
            raise e

    def write_stream(self):
        """
        Writes streaming Process Control Data Model data to Delta using foreachBatch
        """
        try:
            TRIGGER_OPTION = (
                {"availableNow": True}
                if self.trigger == "availableNow"
                else {"processingTime": self.trigger}
            )
            if self.merge == True:
                query = (
                    self.data.writeStream.trigger(**TRIGGER_OPTION)
                    .format("delta")
                    .foreachBatch(self._write_stream_microbatches)
                    .queryName(self.query_name)
                    .outputMode("update")
                    .options(**self.options)
                    .start()
                )
            else:
                delta_float = SparkDeltaDestination(
                    data=self.data.select("TagName", "EventTime", "Status", "Value")
                    .filter(ValueTypeConstants.FLOAT_VALUE)
                    .withColumn("Value", col("Value").cast("float")),
                    destination=self.destination_float,
                    options=self.options,
                    mode=self.mode,
                    trigger=self.trigger,
                    query_name=self.query_name + "_float",
                )

                delta_float.write_stream()

                if self.destination_string != None:
                    delta_string = SparkDeltaDestination(
                        data=self.data.select(
                            "TagName", "EventTime", "Status", "Value"
                        ).filter(ValueTypeConstants.STRING_VALUE),
                        destination=self.destination_string,
                        options=self.options,
                        mode=self.mode,
                        trigger=self.trigger,
                        query_name=self.query_name + "_string",
                    )

                    delta_string.write_stream()

                if self.destination_integer != None:
                    delta_integer = SparkDeltaDestination(
                        data=self.data.select(
                            "TagName", "EventTime", "Status", "Value"
                        ).filter(ValueTypeConstants.INTEGER_VALUE),
                        destination=self.destination_integer,
                        options=self.options,
                        mode=self.mode,
                        trigger=self.trigger,
                        query_name=self.query_name + "_integer",
                    )

                    delta_integer.write_stream()

                while self.spark.streams.active != []:
                    for query in self.spark.streams.active:
                        if query.lastProgress:
                            logging.info(
                                "{}: {}".format(query.name, query.lastProgress)
                            )
                    time.sleep(10)

        except Py4JJavaError as e:
            logging.exception(e.errmsg)
            raise e
        except Exception as e:
            logging.exception(str(e))
            raise e

system_type() staticmethod

Attributes:

Name	Type	Description
SystemType	Environment	Requires PYSPARK

Source code in src/sdk/python/rtdip_sdk/pipelines/destinations/spark/pcdm_to_delta.py

@staticmethod
def system_type():
    """
    Attributes:
        SystemType (Environment): Requires PYSPARK
    """
    return SystemType.PYSPARK

write_batch()

Writes Process Control Data Model data to Delta

Source code in src/sdk/python/rtdip_sdk/pipelines/destinations/spark/pcdm_to_delta.py

def write_batch(self):
    """
    Writes Process Control Data Model data to Delta
    """
    try:
        if self.try_broadcast_join != True:
            self.data.persist()

        self._write_data_by_type(self.data)

        if self.try_broadcast_join != True:
            self.data.unpersist()

    except Py4JJavaError as e:
        logging.exception(e.errmsg)
        raise e
    except Exception as e:
        logging.exception(str(e))
        raise e

write_stream()

Writes streaming Process Control Data Model data to Delta using foreachBatch

Source code in src/sdk/python/rtdip_sdk/pipelines/destinations/spark/pcdm_to_delta.py

def write_stream(self):
    """
    Writes streaming Process Control Data Model data to Delta using foreachBatch
    """
    try:
        TRIGGER_OPTION = (
            {"availableNow": True}
            if self.trigger == "availableNow"
            else {"processingTime": self.trigger}
        )
        if self.merge == True:
            query = (
                self.data.writeStream.trigger(**TRIGGER_OPTION)
                .format("delta")
                .foreachBatch(self._write_stream_microbatches)
                .queryName(self.query_name)
                .outputMode("update")
                .options(**self.options)
                .start()
            )
        else:
            delta_float = SparkDeltaDestination(
                data=self.data.select("TagName", "EventTime", "Status", "Value")
                .filter(ValueTypeConstants.FLOAT_VALUE)
                .withColumn("Value", col("Value").cast("float")),
                destination=self.destination_float,
                options=self.options,
                mode=self.mode,
                trigger=self.trigger,
                query_name=self.query_name + "_float",
            )

            delta_float.write_stream()

            if self.destination_string != None:
                delta_string = SparkDeltaDestination(
                    data=self.data.select(
                        "TagName", "EventTime", "Status", "Value"
                    ).filter(ValueTypeConstants.STRING_VALUE),
                    destination=self.destination_string,
                    options=self.options,
                    mode=self.mode,
                    trigger=self.trigger,
                    query_name=self.query_name + "_string",
                )

                delta_string.write_stream()

            if self.destination_integer != None:
                delta_integer = SparkDeltaDestination(
                    data=self.data.select(
                        "TagName", "EventTime", "Status", "Value"
                    ).filter(ValueTypeConstants.INTEGER_VALUE),
                    destination=self.destination_integer,
                    options=self.options,
                    mode=self.mode,
                    trigger=self.trigger,
                    query_name=self.query_name + "_integer",
                )

                delta_integer.write_stream()

            while self.spark.streams.active != []:
                for query in self.spark.streams.active:
                    if query.lastProgress:
                        logging.info(
                            "{}: {}".format(query.name, query.lastProgress)
                        )
                time.sleep(10)

    except Py4JJavaError as e:
        logging.exception(e.errmsg)
        raise e
    except Exception as e:
        logging.exception(str(e))
        raise e