spark read a file based on value in Dataframe

Kallie 0
Viewed 28

I'm reading messages from kafka. The messages schema is -

schema = StructType([
    StructField("file_path", StringType(), True),
    StructField("table_name", StringType(), True),
])

For each row in the dataframe that I read, I want to open the file in the specified file_path, and write it to a delta lake table with the same name as in the column table_name.

So for example if the row in the dataframe is -

-------------------------------
|   file_path   | table_name  |
-------------------------------
| /tmp/file.csv |   table_1   |
-------------------------------

I want to be able to do -

data = spark.read.csv(df["file_path"])
data.write.format("delta").mode("append").saveAsTable(df["table_name"])
delta-lake pyspark apache-spark python
1 Answers

Once you receive the data stream from kafka, maybe you can try something like this.

dataframe.foreach(rowObject =>{
// get row object
// have your logic to read the rowObject(0) which has the file path and write to table rowObject(1) which has table name
data = spark.read.csv(rowObject(0))
data.write.format("delta").mode("append").saveAsTable("rowObject(1)")
})
Shane0
Related