tencent cloud

This example demonstrates how to use Spark Streaming to read the reasoning table in COS storage in streaming mode, process key fields (feature, predicted value), and write to the DLC internal table.

## Prerequisite
Ensure the COS access credential associated with the DLC engine has permission to access the reasoning table storage bucket.
2. Underwrite the model service has enabled reasoning table monitoring.

## How to run
After modifying parameters in the tutorial, select DLC PySpark task from offline development > orchestration space to run.

## Scheduled job

After modifying the corresponding parameters and completing debugging, you can configure the task for scheduled execution to continuously process reasoning table data. You can also associate the processed reasoning table data with tag data to generate a reasoning table containing tag data, and configure machine learning-related monitoring metrics for the reasoning table in the data quality module to continuously monitor model quality.
"""from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F
from pyspark.sql import types as T
from typing import Union, List, Dict, Optional, Sequence, Any
from pyspark.sql.streaming import StreamingQuery

from pyspark.sql.types import (
    StructType, StructField, StringType, LongType, DoubleType,
    ArrayType, TimestampType
)

"""
### Define original reasoning table data format
The original reasoning table data is stored in the user-specified COS bucket in JSON format with a specific structure. request_schema represents the input request structure, and response_schema represents the model output structure. Adjust according to the actual input and output format of the model service. Other fields are part of the fixed structure of the reasoning table and generally not needed to adjust.
"""
request_schema = StructType([
    StructField("client_request_id", StringType(), True),
    StructField("dataframe_split", StructType([
        StructField("columns", ArrayType(StringType()), True),
        StructField("data", ArrayType(ArrayType(DoubleType())), True),
    ]), True),
])

request_metadata_schema = StructType([
    StructField("model_name", StringType(), True),
    StructField("model_version", StringType(), True),
    StructField("service_id", StringType(), True),
])

response_schema = StructType([
    StructField("predictions", ArrayType(LongType()), True),
])

schema = StructType([
    StructField("client_request_id", StringType(), True),
    StructField("execution_duration_ms", LongType(), True),
    StructField("request", request_schema, True),
    StructField("request_date", StringType(), True),
    StructField("request_metadata", request_metadata_schema, True),
    StructField("request_time", TimestampType(), True),
    StructField("requester", StringType(), True),
    StructField("response", response_schema, True),
    StructField("sampling_fraction", DoubleType(), True),
    StructField("status_code", LongType(), True),
    StructField("wedata_request_id", StringType(), True),
])

# data source parameters
SOURCE_PATH = "cosn://bucket path/"  # storage path of the reasoning table
CHECKPOINT_PATH = "cosn://bucket path/checkpoint" # storage path of check_point, recommended to match the reasoning table storage path

# result table parameters
UNPACKED_TABLE_NAME = "DataLakeCatalog.test.unpacked_test_inference_table"  # Result table name, need to be a three-segment table name in DLC: <data catalog>.<database>.<data table>
MODEL_ID_COL = "model_id" # model ID column in the result table, used for identification model
EXAMPLE_ID_COL = "example_id" # unique ID column for result table entries, used for update operations
PREDICTION_COL = "prediction" # column name for predicted value
FEATURE_COLUMNS = ["sepal_length", "sepal_width", "petal_length", "petal_width"] # feature column names, adjust based on model features


def process_requests(requests_raw: DataFrame) -> DataFrame:
    """
    Expand request features into separate columns and pair them with prediction results

    Args:
        requests_raw: pending data
    Return:
        processed data
    """
    filter successful requests
    requests_success = requests_raw.filter(F.col("status_code") == 200).drop("status_code")

    # Generate model flag
    requests_identified = requests_success \\
        .withColumn(MODEL_ID_COL, F.concat(F.col("request_metadata").getItem("model_name"), F.lit("_"), F.col("request_metadata").getItem("model_version"))) \\
        .drop("request_metadata")

    # Unfold feature column and prediction result
    # 1. Get feature column names, feature data, predicted value
    requests_with_features = (
        requests_identified
        .withColumn("feature_columns", F.col("request.dataframe_split.columns"))
        .withColumn("feature_data", F.col("request.dataframe_split.data"))
        .withColumn("predictions", F.col("response.predictions"))
    )

    # 2. Pair each data row with the corresponding prediction result
    requests_exploded = (
        requests_with_features
        .withColumn("feature_prediction_pairs", 
                   F.arrays_zip(F.col("feature_data"), F.col("predictions")))
        .withColumn("feature_prediction_pairs", 
                   F.explode(F.col("feature_prediction_pairs")))
        .withColumn("feature_row", F.col("feature_prediction_pairs.feature_data"))
        .withColumn(PREDICTION_COL, F.col("feature_prediction_pairs.predictions"))
    )

    # 3. dynamically create feature column
    requests_with_feature_cols = requests_exploded
    for i, col_name in enumerate(FEATURE_COLUMNS):
        requests_with_feature_cols = (
            requests_with_feature_cols
            .withColumn(col_name, F.col("feature_row")[i])
        )
    
    # 4. Add record unique ID, making it easy to perform upsert operation
    requests_with_example_id = requests_with_feature_cols \\
        .withColumn(EXAMPLE_ID_COL, F.expr("uuid()"))

    # 5. Clean up temporary columns
    requests_processed = (
        requests_with_example_id
        .drop("feature_columns", "feature_data", "feature_prediction_pairs", "feature_row", "predictions", "request", "response")
    )

    return requests_processed


def create_table(
        spark: SparkSession,
        table_name: str,
        df: Optional[DataFrame] = None,
        partition_expr: str = None,
        description: Optional[str] = None,
):

    """
    Create a dlc internal table

    Args:
        spark: SparkSession instance
        table_name: full table name (format: data catalog.database.data table)
        df: initial data (used for infer schema)
        partition_expr: partition column (optimized storage query)
        description: table description
    Raises:
        ValueError: throw create exception
    """
    # Infer table schema
    table_schema = df.schema
    
    # Build column definition
    columns_ddl = []
    for field in table_schema.fields:
        data_type = field.dataType.simpleString().upper()
        col_def = f"`{field.name}` {data_type}"
        if not field.nullable:
            col_def += " NOT NULL"
        columns_ddl.append(col_def)
    
    # Build table creation statement
    ddl = f"""
    CREATE TABLE IF NOT EXISTS {table_name} (
        {', '.join(columns_ddl)}
    )
    USING iceberg
    PARTITIONED BY ({partition_expr})
    TBLPROPERTIES (
        'comment'= '{description or ''}',
        'format-version'= '2',
        'write.metadata.previous-versions-max'= '100',
        'write.metadata.delete-after-commit.enabled'= 'true',
        'smart-optimizer.inherit' = 'none',
        'smart-optimizer.written.enable' = 'enable'
    )
    """

    # Print sql
    print(f"create table ddl: {ddl}\\n")

    # execute DDL
    try:
        spark.sql(ddl)
    except Exception as e:
        raise ValueError(f"Failed to create table: {str(e)}") from e

    print(f"create table {table_name} done")

spark = SparkSession.builder.appName("Operate DB Example").getOrCreate()

# Build a streaming DataFrame
df = (
    spark.readStream.format("json")
    .schema(schema)
    .load(SOURCE_PATH)
)

# Process streaming data
unpacked_df = process_requests(df)

# Create a result table with request_time as partition field, using daily partitioning, which can be adjusted by service request volume
create_table(spark, UNPACKED_TABLE_NAME, unpacked_df, 'days(request_time)')

# Write data
query = unpacked_df.writeStream \\
        .format("parquet") \\
        .outputMode("append") \\
        .option("checkpointLocation", CHECKPOINT_PATH) \\
        .trigger(once=True) \\
        .toTable(UNPACKED_TABLE_NAME)

query.awaitTermination()  


spark.sql(f"select * from {UNPACKED_TABLE_NAME}").show(10)



"""
This example demonstrates how to use Spark Streaming to read the reasoning table from COS storage in streaming mode, process key fields (feature, predicted value), and write to the EMR Hive table.

## Prerequisite
Ensure the COS access credential associated with the DLC engine has permission to access the reasoning table storage bucket.
2. Underwrite the model service has enabled reasoning table monitoring.

## How to run
After modifying parameters in the tutorial, select EMR PySpark task from offline development > orchestration space to run.

## Scheduled job

After modifying the corresponding parameters and completing debugging, you can configure the task for scheduled execution to continuously process reasoning table data. You can also associate the processed reasoning table data with tag data to generate a reasoning table containing tag data, and configure machine learning-related monitoring metrics for the reasoning table in the data quality module to continuously monitor model quality.

"""
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F
from pyspark.sql import types as T
from typing import Union, List, Dict, Optional, Sequence, Any
from pyspark.sql.streaming import StreamingQuery

from pyspark.sql.types import (
    StructType, StructField, StringType, LongType, DoubleType,
    ArrayType, TimestampType
)

"""
### Define original reasoning table data format
The original reasoning table data is stored in the user-specified COS bucket in JSON format with a specific structure. request_schema represents the input request structure, and response_schema represents the model output structure. Adjust according to the actual input and output format of the model service. Other fields are part of the fixed structure of the reasoning table and generally not needed to adjust.
"""

request_schema = StructType([
    StructField("client_request_id", StringType(), True),
    StructField("dataframe_split", StructType([
        StructField("columns", ArrayType(StringType()), True),
        StructField("data", ArrayType(ArrayType(DoubleType())), True),
    ]), True),
])

request_metadata_schema = StructType([
    StructField("model_name", StringType(), True),
    StructField("model_version", StringType(), True),
    StructField("service_id", StringType(), True),
])

response_schema = StructType([
    StructField("predictions", ArrayType(LongType()), True),
])

schema = StructType([
    StructField("client_request_id", StringType(), True),
    StructField("execution_duration_ms", LongType(), True),
    StructField("request", request_schema, True),
    StructField("request_date", StringType(), True),
    StructField("request_metadata", request_metadata_schema, True),
    StructField("request_time", TimestampType(), True),
    StructField("requester", StringType(), True),
    StructField("response", response_schema, True),
    StructField("sampling_fraction", DoubleType(), True),
    StructField("status_code", LongType(), True),
    StructField("wedata_request_id", StringType(), True),
])


# data source parameters
SOURCE_PATH = "cosn://bucket path/"  # storage path of the reasoning table
CHECKPOINT_PATH = "cosn://bucket path/checkpoint" # storage path of check_point, recommended to match the reasoning table storage path

# result table parameters
UNPACKED_TABLE_NAME = "testdb.UNPACKED_test_inference_TABLE"  # Result TABLE NAME, need to be a hive TABLE NAME: <database>.<data TABLE>
MODEL_ID_COL = "model_id" # model ID column in the result table, used for identification model
EXAMPLE_ID_COL = "example_id" # unique ID column for result table entries, used for update operations
PREDICTION_COL = "prediction" # column name for predicted value
FEATURE_COLUMNS = ["sepal_length", "sepal_width", "petal_length", "petal_width"] # feature column names, adjust based on model features

def process_requests(requests_raw: DataFrame) -> DataFrame:
    """
    Expand request features into separate columns and pair them with prediction results

    Args:
        requests_raw: pending data
    Return:
        processed data
    """
    filter successful requests
    requests_success = requests_raw.filter(F.col("status_code") == 200).drop("status_code")

    # Generate model flag
    requests_identified = requests_success \\
        .withColumn(MODEL_ID_COL, F.concat(F.col("request_metadata").getItem("model_name"), F.lit("_"), F.col("request_metadata").getItem("model_version"))) \\
        .drop("request_metadata")

    # Unfold feature column and prediction result
    # 1. Get feature column names, feature data, predicted value
    requests_with_features = (
        requests_identified
        .withColumn("feature_columns", F.col("request.dataframe_split.columns"))
        .withColumn("feature_data", F.col("request.dataframe_split.data"))
        .withColumn("predictions", F.col("response.predictions"))
    )

    # 2. Pair each data row with the corresponding prediction result
    requests_exploded = (
        requests_with_features
        .withColumn("feature_prediction_pairs", 
                   F.arrays_zip(F.col("feature_data"), F.col("predictions")))
        .withColumn("feature_prediction_pairs", 
                   F.explode(F.col("feature_prediction_pairs")))
        .withColumn("feature_row", F.col("feature_prediction_pairs.feature_data"))
        .withColumn(PREDICTION_COL, F.col("feature_prediction_pairs.predictions"))
    )

    # 3. dynamically create feature column
    requests_with_feature_cols = requests_exploded
    for i, col_name in enumerate(FEATURE_COLUMNS):
        requests_with_feature_cols = (
            requests_with_feature_cols
            .withColumn(col_name, F.col("feature_row")[i])
        )
    
    # 4. Add record unique ID, making it easy to perform upsert operation
    requests_with_example_id = requests_with_feature_cols \\
        .withColumn(EXAMPLE_ID_COL, F.expr("uuid()"))

    # 5. Clean up temporary columns
    requests_processed = (
        requests_with_example_id
        .drop("feature_columns", "feature_data", "feature_prediction_pairs", "feature_row", "predictions", "request", "response")
    )

    return requests_processed


def create_table(
        spark: SparkSession,
        table_name: str,
        df: Optional[DataFrame] = None,
        partition_expr: str = None,
        description: Optional[str] = None,
):
    """
    Create a Hive table.

    Args:
        spark: SparkSession instance
        table_name: full hive table name (format: database.data table)
        df: initial data (optional, used for infer schema)
        partition_expr: partition column (optimized storage query)
        description: table description
    Raises:
        ValueError: throw create exception
    """
    # Infer table schema
    table_schema = df.schema

    part_col_nam = partition_expr.split(' ')[0]
    
    # Build column definition
    columns_ddl = []
    for field in table_schema.fields:
        # Filter out partition columns when creating a table
        if field.name == part_col_nam:
            continue
        data_type = field.dataType.simpleString().upper()
        col_def = f"`{field.name}` {data_type}"
        if not field.nullable:
            col_def += " NOT NULL"
        columns_ddl.append(col_def)
    
    # Build table creation statement
    ddl = f"""
    CREATE TABLE IF NOT EXISTS {table_name} (
        {', '.join(columns_ddl)}
    )
    PARTITIONED BY ({partition_expr})
    TBLPROPERTIES (
        'comment'= '{description or ''}'
    )
    """
    
    # Print sql
    print(f"create table ddl: {ddl}\\n")

    # execute DDL
    try:
        spark.sql(ddl)
    except Exception as e:
        raise ValueError(f"Failed to create table: {str(e)}") from e

    print(f"create table {table_name} done")


def write_batch_to_hive(batch_df: DataFrame, batch_id: int, table_name: str):
    # Control small files: repartition by partition column
    out_df = batch_df.repartition(8, "dt")
    
    out_df.write.mode("append").format("hive").saveAsTable(
        table_name,
        partitionBy="dt"
    )

spark = SparkSession.builder.appName("Operate DB Example").enableHiveSupport().getOrCreate()

# Build a streaming DataFrame
df = (
    spark.readStream.format("json")
    .schema(schema)
    .load(SOURCE_PATH)
)

# Process streaming data
unpacked_df = process_requests(df)

# Insert partition column
requests_with_dt = (
        unpacked_df
        .withColumn("dt", F.substring(F.col("request_date"), 1, 10))  # Extract date in YYYY-MM-DD format
)

# Create a result table
create_table(spark, UNPACKED_TABLE_NAME, unpacked_df, 'dt STRING')

# Set Hive dynamic partition
spark.sql("SET hive.exec.dynamic.partition=true")
spark.sql("SET hive.exec.dynamic.partition.mode=nonstrict")

# Batch write to result table
query = (
    requests_with_dt
    .writeStream
    .outputMode("append")
    .option("checkpointLocation", CHECKPOINT_PATH)
    .trigger(once=True)
    .foreachBatch(lambda df, id: write_batch_to_hive(df, id, UNPACKED_TABLE_NAME))
    .start()
)

query.awaitTermination()

# Query result data
spark.sql(f"select * from {UNPACKED_TABLE_NAME}").show(10)