I want to create a streaming job, that reads messages from a folder within TXT files, does the parsing, some processing, and appends the result into one of 3 possible delta tables depending on the parse result. There is a parse_failed table, an unknwon_msgs table, and a parsed_msgs table.
Reading is done with
sdf = spark.readStream.text(path=path_input, lineSep="\n\n", pathGlobFilter="*.txt", recursiveFileLookup=True)
and writing with
x = sdf.writeStream.foreachBatch(process_microbatch).start()
where process_microbatch is
def process_microbatch(self, batch_df: DataFrame, batch_id: int) -> None:
"""Processing of newly arrived messages. For each message replicate it if needed, and execute the parse_msg_proxy on each."""
batch_df.rdd.flatMap(lambda msg: replicate_msg(msg)).map(lambda msg: parse_msg_proxy(msg))
and where parse_msg_proxy is
def parse_msg_proxy(self, msg: str) -> None:
try:
parsed_msg = parse_message(msg, element_mapping)
# do some processing
# create df_msg dataframe from parsed_msg
df_msg.write.format("delta").mode("append").save(path_parsed_msgs)
except ParseException as e:
spark.createDataFrame([{'msg': parsed_msg, 'error': str(e)}]).write.format("delta").mode("append").save(path_parse_errors)
raise Exception("Parse error occured.")
except UnknownMsgTypeException:
spark.createDataFrame([{'msg': parsed_msg}]).write.format("delta").mode("append").save(path_unknown_msgs)
The streaming job starts without error message, but the delta tables are not created. Whats wrong? Maybe an action operations are missing, like collect()? Where should I put it?
