Skip to content

Convert SSIP PI Binary JSON data to the Process Control Data Model

SSIPPIJsonStreamToPCDMTransformer

Bases: TransformerInterface

Converts a Spark DataFrame containing Binary JSON data and related Properties to the Process Control Data Model

For more information about the SSIP PI Streaming Connector, please see here.

Example

from rtdip_sdk.pipelines.transformers import SSIPPIJsonStreamToPCDMTransformer
from rtdip_sdk.pipelines.utilities import SparkSessionUtility

# Not required if using Databricks
spark = SparkSessionUtility(config={}).execute()

ssip_pi_json_stream_to_pcdm_transformer = SSIPPIJsonStreamToPCDMTransformer(
    spark=spark,
    data=df,
    source_column_name="body",
    properties_column_name="",
    metadata_delta_table=None
)

result = ssip_pi_json_stream_to_pcdm_transformer.transform()

Parameters:

Name Type Description Default
spark SparkSession

Spark Session

 required
data DataFrame

DataFrame containing the path and binaryFile data

 required
source_column_name str

Spark Dataframe column containing the Binary json data

 required
properties_column_name str

Spark Dataframe struct typed column containing an element with the PointType

 required
metadata_delta_table (optional, str)

Name of a metadata table that can be used for PointType mappings

 None
Source code in src/sdk/python/rtdip_sdk/pipelines/transformers/spark/ssip_pi_binary_json_to_pcdm.py 
 23
 24
 25
 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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
class SSIPPIJsonStreamToPCDMTransformer(TransformerInterface):
    """
    Converts a Spark DataFrame containing Binary JSON data and related Properties to the Process Control Data Model

    For more information about the SSIP PI Streaming Connector, please see [here.](https://bakerhughesc3.ai/oai-solution/shell-sensor-intelligence-platform/)

    Example
    --------
    ```python
    from rtdip_sdk.pipelines.transformers import SSIPPIJsonStreamToPCDMTransformer
    from rtdip_sdk.pipelines.utilities import SparkSessionUtility

    # Not required if using Databricks
    spark = SparkSessionUtility(config={}).execute()

    ssip_pi_json_stream_to_pcdm_transformer = SSIPPIJsonStreamToPCDMTransformer(
        spark=spark,
        data=df,
        source_column_name="body",
        properties_column_name="",
        metadata_delta_table=None
    )

    result = ssip_pi_json_stream_to_pcdm_transformer.transform()
    ```

    Parameters:
        spark (SparkSession): Spark Session
        data (DataFrame): DataFrame containing the path and binaryFile data
        source_column_name (str): Spark Dataframe column containing the Binary json data
        properties_column_name (str): Spark Dataframe struct typed column containing an element with the PointType
        metadata_delta_table (optional, str): Name of a metadata table that can be used for PointType mappings
    """

    spark: SparkSession
    data: DataFrame
    source_column_name: str
    properties_column_name: str
    metadata_delta_table: str

    def __init__(
        self,
        spark: SparkSession,
        data: DataFrame,
        source_column_name: str,
        properties_column_name: str,
        metadata_delta_table: str = None,
    ) -> None:
        self.spark = spark
        self.data = data
        self.source_column_name = source_column_name
        self.properties_column_name = properties_column_name
        self.metadata_delta_table = metadata_delta_table

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

    @staticmethod
    def libraries():
        libraries = Libraries()
        return libraries

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

    def pre_transform_validation(self):
        return True

    def post_transform_validation(self):
        return True

    def transform(self) -> DataFrame:
        """
        Returns:
            DataFrame: A dataframe with the provided Binary data converted to PCDM
        """
        df = (
            self.data.withColumn(
                self.source_column_name, col(self.source_column_name).cast("string")
            )
            .withColumn(
                "EventDate",
                get_json_object(col(self.source_column_name), "$.EventTime").cast(
                    "date"
                ),
            )
            .withColumn(
                "TagName",
                get_json_object(col(self.source_column_name), "$.TagName").cast(
                    "string"
                ),
            )
            .withColumn(
                "EventTime",
                get_json_object(col(self.source_column_name), "$.EventTime").cast(
                    "timestamp"
                ),
            )
            .withColumn(
                "Status",
                get_json_object(col(self.source_column_name), "$.Quality").cast(
                    "string"
                ),
            )
            .withColumn(
                "Value",
                get_json_object(col(self.source_column_name), "$.Value").cast("string"),
            )
            .withColumn(
                "PointType", element_at(col(self.properties_column_name), "PointType")
            )
            .withColumn(
                "Action",
                element_at(col(self.properties_column_name), "Action").cast("string"),
            )
        )

        if self.metadata_delta_table != None:
            metadata_df = SparkDeltaSource(
                self.spark, {}, self.metadata_delta_table
            ).read_batch()
            metadata_df = metadata_df.select(
                "TagName", col("PointType").alias("MetadataPointType")
            )
            df = df.join(metadata_df, (df.TagName == metadata_df.TagName), "left")
            df = df.withColumn(
                "PointType",
                (when(col("PointType").isNull(), col("MetadataPointType"))).otherwise(
                    col("PointType")
                ),
            )

        return (
            df.withColumn(
                "ValueType",
                (
                    when(col("PointType") == "Digital", "string")
                    .when(col("PointType") == "String", "string")
                    .when(col("PointType") == "Float16", "float")
                    .when(col("PointType") == "Float32", "float")
                    .when(col("PointType") == "Float64", "float")
                    .when(col("PointType") == "Int16", "integer")
                    .when(col("PointType") == "Int32", "integer")
                    .otherwise("string")
                ),
            )
            .selectExpr(
                "*",
                "CASE WHEN ValueType = 'integer' THEN try_cast(Value as integer) END as Value_Integer",
                "CASE WHEN ValueType = 'float' THEN try_cast(Value as float) END as Value_Float",
            )
            .withColumn(
                "ValueType",
                when(
                    (col("Value_Integer").isNull()) & (col("ValueType") == "integer"),
                    "string",
                )
                .when(
                    (col("Value_Float").isNull()) & (col("ValueType") == "float"),
                    "string",
                )
                .otherwise(col("ValueType")),
            )
            .withColumn(
                "ChangeType",
                (
                    when(col("Action") == "Insert", "insert")
                    .when(col("Action") == "Add", "insert")
                    .when(col("Action") == "Delete", "delete")
                    .when(col("Action") == "Update", "update")
                    .when(col("Action") == "Refresh", "update")
                ),
            )
            .select(
                col("EventDate"),
                col("TagName"),
                col("EventTime"),
                col("Status"),
                col("Value"),
                col("ValueType"),
                col("ChangeType"),
            )
        )

system_type() staticmethod

Attributes:

Name Type Description
SystemType Environment

Requires PYSPARK
Source code in src/sdk/python/rtdip_sdk/pipelines/transformers/spark/ssip_pi_binary_json_to_pcdm.py 
77
78
79
80
81
82
83
@staticmethod
def system_type():
    """
    Attributes:
        SystemType (Environment): Requires PYSPARK
    """
    return SystemType.PYSPARK

transform()

Returns:

Name Type Description
DataFrame DataFrame

A dataframe with the provided Binary data converted to PCDM
Source code in src/sdk/python/rtdip_sdk/pipelines/transformers/spark/ssip_pi_binary_json_to_pcdm.py 
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def transform(self) -> DataFrame:
    """
    Returns:
        DataFrame: A dataframe with the provided Binary data converted to PCDM
    """
    df = (
        self.data.withColumn(
            self.source_column_name, col(self.source_column_name).cast("string")
        )
        .withColumn(
            "EventDate",
            get_json_object(col(self.source_column_name), "$.EventTime").cast(
                "date"
            ),
        )
        .withColumn(
            "TagName",
            get_json_object(col(self.source_column_name), "$.TagName").cast(
                "string"
            ),
        )
        .withColumn(
            "EventTime",
            get_json_object(col(self.source_column_name), "$.EventTime").cast(
                "timestamp"
            ),
        )
        .withColumn(
            "Status",
            get_json_object(col(self.source_column_name), "$.Quality").cast(
                "string"
            ),
        )
        .withColumn(
            "Value",
            get_json_object(col(self.source_column_name), "$.Value").cast("string"),
        )
        .withColumn(
            "PointType", element_at(col(self.properties_column_name), "PointType")
        )
        .withColumn(
            "Action",
            element_at(col(self.properties_column_name), "Action").cast("string"),
        )
    )

    if self.metadata_delta_table != None:
        metadata_df = SparkDeltaSource(
            self.spark, {}, self.metadata_delta_table
        ).read_batch()
        metadata_df = metadata_df.select(
            "TagName", col("PointType").alias("MetadataPointType")
        )
        df = df.join(metadata_df, (df.TagName == metadata_df.TagName), "left")
        df = df.withColumn(
            "PointType",
            (when(col("PointType").isNull(), col("MetadataPointType"))).otherwise(
                col("PointType")
            ),
        )

    return (
        df.withColumn(
            "ValueType",
            (
                when(col("PointType") == "Digital", "string")
                .when(col("PointType") == "String", "string")
                .when(col("PointType") == "Float16", "float")
                .when(col("PointType") == "Float32", "float")
                .when(col("PointType") == "Float64", "float")
                .when(col("PointType") == "Int16", "integer")
                .when(col("PointType") == "Int32", "integer")
                .otherwise("string")
            ),
        )
        .selectExpr(
            "*",
            "CASE WHEN ValueType = 'integer' THEN try_cast(Value as integer) END as Value_Integer",
            "CASE WHEN ValueType = 'float' THEN try_cast(Value as float) END as Value_Float",
        )
        .withColumn(
            "ValueType",
            when(
                (col("Value_Integer").isNull()) & (col("ValueType") == "integer"),
                "string",
            )
            .when(
                (col("Value_Float").isNull()) & (col("ValueType") == "float"),
                "string",
            )
            .otherwise(col("ValueType")),
        )
        .withColumn(
            "ChangeType",
            (
                when(col("Action") == "Insert", "insert")
                .when(col("Action") == "Add", "insert")
                .when(col("Action") == "Delete", "delete")
                .when(col("Action") == "Update", "update")
                .when(col("Action") == "Refresh", "update")
            ),
        )
        .select(
            col("EventDate"),
            col("TagName"),
            col("EventTime"),
            col("Status"),
            col("Value"),
            col("ValueType"),
            col("ChangeType"),
        )
    )