Converting CSV to JSON using pyspark with filter and additional columns

user_10951110 0
Viewed 101

Given a CSV Input file with Headers:

"CorrelationID", "Message", "EventTimeStamp", "Flag", "RandomColumns"
12345, "Hello", "2019-06-09 04:25:15", "True", "blah"
12345, "Hello", "2019-06-09 04:25:18", "False", "blah"
45678, "Brick", "2019-06-09 04:26:23", "True", "blah"
78912, "Stone", "2019-06-09 04:29:50", "False", "blah"

Consider only those CorrelationID which has both true and false Flag. Ignore the rest of rows which don't contain both "true" and "false" value for "flag" column

EventTimeStamp value for True flag goes as StartTime and EventTimeStamp value for False flag goes as EndTime

JSON file format as Output:

{"CorrelationID": "12345","Message":"Hello","StartTime":"2019-06-09 04:25:15","EndTime":"2019-06-09 04:25:18"}
pyspark-dataframes pyspark apache-spark-sql join json
1 Answers

Use groupBy & inside agg function use collect_set,first & last functions. Check below code.

from pyspark.sql import functions as F

df \
.withColumn(\ # casting eventtimestamp to timestamp
    "eventtimestamp", \
    F.col("eventtimestamp").cast("timestamp")\
) \
.orderBy(F.col("eventtimestamp").asc) \ # sorting eventtimestamp asc
.groupBy(F.col("correlationid"),F.col("Message")) \ # grouping records based on correlationid
.agg( \
    F.first(F.col("eventtimestamp")).cast("string").alias("StartTime"),\ # First value of eventtimestamp as StartTime
    F.last(F.col("eventtimestamp")).cast("string").alias("EndTime"), \ # Last value of eventtimestamp as End Time
    F.collect_set(F.col("flag")).alias("flag")\ # Collecting Set Of flags & Use size of this value in filter to get only records which has true and false for correlationid.
) \
.filter(F.size(F.col("flag")) === 2) \ 
.select( \
    F.to_json(\ # Adding required columns to inside struct to make json record
        F.struct(\
            F.col("CorrelationID"),\
            F.col("Message"), \
            F.col("StartTime"), \
            F.col("EndTime") \
        ).alias("json_data")\
    ) \
) \
.show(false)

+-------------------------------------------------------------------------------------------------------------+
|json_data                                                                                                    |
+-------------------------------------------------------------------------------------------------------------+
|{"CorrelationID":"12345","Message":"Hello","StartTime":"2019-06-09 04:25:15","EndTime":"2019-06-09 04:25:18"}|
+-------------------------------------------------------------------------------------------------------------+
user_85934140
Related