Hi @SuMiT1 ,

So you can try below approach:

1. First prepare sample data that reflects your json content

from pyspark.sql.functions import col, explode, from_json, get_json_object
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, IntegerType


sample_data = [
    {
        "conversation_id": "conv_001",
        "session_id": "sess_12345",
        "user_id": "user_001",
        "timestamp": "2024-10-09T10:30:00",
        "content": '{"activities": [{"valueType": "ConversationInfo", "type": "trace", "timestamp": 1758964295, "from": {"id": "", "role": 0}, "value": {"lastSessionOutcome": "Abandoned", "lastSessionOutcomeReason": "UserExit", "isDesignMode": true, "locale": "en-US"}}, {"id": "a96558ddb", "type": "event", "timestamp": 1758964295, "from": {"id": "a0899dd8e7", "role": 1}, "name": "startConversation", "channelId": "pio"}]}',
        "metadata": '{"source": "web", "channel": "chat", "region": "US", "deviceType": "desktop", "browserInfo": {"name": "Chrome", "version": "120"}}'
    },
    {
        "conversation_id": "conv_002",
        "session_id": "sess_67890",
        "user_id": "user_002",
        "timestamp": "2024-10-09T11:15:00",
        "content": '{"activities": [{"valueType": "VariableAssignment", "id": "7ce594904b4af48", "type": "trace", "timestamp": 1758964296, "from": {"id": "e234528bb5cb3", "role": 0}, "value": {"name": "LoggedInUserEmail", "id": "Global.LoggedInUserEmail", "newValue": "test@xyz.com", "type": "global"}}]}',
        "metadata": '{"source": "mobile", "channel": "app", "region": "EU", "deviceType": "smartphone", "appVersion": "2.5.1"}'
    }
]

df = spark.createDataFrame(sample_data)

2. Define schemas - needed for from_json function:

activity_schema = StructType([
    StructField("valueType", StringType(), True),
    StructField("id", StringType(), True),
    StructField("type", StringType(), True),
    StructField("timestamp", IntegerType(), True),
    StructField("from", StructType([
        StructField("id", StringType(), True),
        StructField("role", IntegerType(), True)
    ]), True),
    StructField("name", StringType(), True),
    StructField("channelId", StringType(), True),
    StructField("text", StringType(), True),
    StructField("value", StringType(), True)  #This can be parser further if you want
])

content_schema = StructType([
    StructField("activities", ArrayType(activity_schema), True)
])


metadata_schema = StructType([
    StructField("source", StringType(), True),
    StructField("channel", StringType(), True),
    StructField("region", StringType(), True),
    StructField("deviceType", StringType(), True),
    StructField("browserInfo", StructType([
        StructField("name", StringType(), True),
        StructField("version", StringType(), True)
    ]), True),
    StructField("appVersion", StringType(), True)
])

3. Parse data using from_json function and schemas defined at step above:

df_complete = (df 
    .withColumn("content_parsed", from_json(col("content"), content_schema)) \
    .withColumn("metadata_parsed", from_json(col("metadata"), metadata_schema)) \
    .withColumn("activity", explode(col("content_parsed.activities"))) \
    .select(
        
        col("conversation_id"),
        col("session_id"),
        col("user_id"),
        col("timestamp").alias("conversation_timestamp"),
        
        # Metadata columns
        col("metadata_parsed.source").alias("meta_source"),
        col("metadata_parsed.channel").alias("meta_channel"),
        col("metadata_parsed.region").alias("meta_region"),
        col("metadata_parsed.deviceType").alias("meta_deviceType"),
        col("metadata_parsed.browserInfo.name").alias("meta_browser_name"),
        col("metadata_parsed.browserInfo.version").alias("meta_browser_version"),
        col("metadata_parsed.appVersion").alias("meta_appVersion"),
        
        # Activity columns
        col("activity.valueType").alias("activity_valueType"),
        col("activity.id").alias("activity_id"),
        col("activity.type").alias("activity_type"),
        col("activity.timestamp").alias("activity_timestamp"),
        col("activity.from.id").alias("from_id"),
        col("activity.from.role").alias("from_role"),
        col("activity.name").alias("activity_name"),
        col("activity.channelId").alias("channel_id"),
        col("activity.text").alias("activity_text"),
        col("activity.value").alias("activity_value")
    )
)