Structured Streaming

sudhanshu1 — Wed, 21 Dec 2022 12:43:11 GMT

I need some solution for below problem.

We have set of json files which are keep coming to aws s3, these files contains details for a property . please note 1 property can have 10-12 rows in this json file. Attached is sample json file.

We need to read these files as stream and then we need to create compacted view of 1 property , which means take all rows for that property and merge and create a single row per property . once that is done , we can write that stream to delta/ documentDB/ DynamoDB.

I have tried below

def mount_s3_bucket(access_key, secret_key, bucket_name, mount_folder):

ACCESS_KEY_ID = access_key

SECRET_ACCESS_KEY = secret_key

ENCODED_SECRET_KEY = SECRET_ACCESS_KEY.replace("/", "%2F")

print ("Mounting", bucket_name)

try:

# Unmount the data in case it was already mounted.

dbutils.fs.unmount("/mnt/%s" % mount_folder)

except:

# If it fails to unmount it most likely wasn't mounted in the first place

print ("Directory not unmounted: ", mount_folder)

finally:

# Lastly, mount our bucket.

dbutils.fs.mount("s3a://%s:%s@%s" % (ACCESS_KEY_ID, ENCODED_SECRET_KEY, bucket_name), "/mnt/%s" % mount_folder)

#dbutils.fs.mount("s3a://"+ ACCESS_KEY_ID + ":" + ENCODED_SECRET_KEY + "@" + bucket_name, mount_folder)

print ("The bucket", bucket_name, "was mounted to", mount_folder, "\n")

# Set AWS programmatic access credentials

from pyspark.sql import SparkSession

from pyspark.sql.functions import *

from pyspark.sql.types import *

spark = SparkSession.builder.appName('Comparis-data-stream-app').getOrCreate()

print('Session created')

# JSONschema = StructType([

# StructField("username",StringType(),True),

# StructField("currency",StringType(),True),

# StructField("amount",LongType(),True),

# ])

JSONschema = StructType([

StructField("id", StringType(), False),

StructField("address1", StringType(), True, None),

StructField("address2", StringType(), True, None),

StructField("city", StringType(), True, None),

StructField("state", StringType(), True, None),

StructField("postalCode", StringType(), True, None),

StructField("price", IntegerType(), True, None),

StructField("size", IntegerType(), True, None),

StructField("rooms", IntegerType(), True, None),

StructField("lat", DecimalType(), True , None),

StructField("lng", DecimalType(), True, None),

StructField("hash_lat_lng", StringType(), True , None),

StructField("source", StringType(), True , None),

StructField("source_premium", StringType(), True, None),

StructField("timestamp", TimestampType(), True , None),

StructField("attributes", StructType([

StructField("type", StringType()),

StructField("lift", StringType()),

StructField("garden", StringType()),

StructField("heating", StringType()),

StructField("washing_machine", StringType()),

StructField("floor", StringType()),

StructField("year_of_construction", IntegerType())

]))])

ds = (spark.readStream

.schema(JSONschema)

.format("json")

.option("maxFilesPerTrigger", 1)

.load("/mnt/raj-zuk-comparis-poc/messages*.json"))

flattened_df = (ds.withColumn("property_type", expr("attributes.type"))

.withColumn("property_lift", expr("attributes.lift"))

.withColumn("property_garden", expr("attributes.garden"))

.withColumn("property_heating", expr("attributes.heating"))

.withColumn("property_washing_machine", expr("attributes.washing_machine"))

.withColumn("property_floor", expr("attributes.floor"))

.withColumn("property_year_of_construction", expr("attributes.year_of_construction"))

)

tumbling_df = (df_a44ed14d77a096c5197933f1e02b7a47

.groupBy(window(col("timestamp"), "4 hour"), col("source"), col("hash_lat_lng"))

.agg(max("timestamp").alias("timestamp"),

first("address1").alias("address1"),

first("address2").alias("address2"),

first("city").alias("city"),

first("state").alias("state"),

first("postalCode").alias("postalCode"),

first("price").alias("price"),

first("size").alias("size"),

first("rooms").alias("rooms"),

first("property_type").alias("property_type"),

first("property_lift").alias("property_lift"),

first("property_garden").alias("property_garden"),

first("property_heating").alias("property_heating"),

first("property_washing_machine").alias("property_washing_machine"),

first("property_floor").alias("property_floor"),

first("property_year_of_construction").alias("property_year_of_construction"),

first("source_premium").alias("source_premium"))

.orderBy(col("window.start"))

)

After this i am not sure how to write this ? could you please advise what should be correct way to write or solve this problem please?

topic Structured Streaming in Data Engineering

Structured Streaming