How to write result to DB in spark map function?

Viewed 22

I have eg.3 files in /home/elogs3/ Now I need to use Py Spark2.4 to resolve each file and write each result to MySQL like a wordcount of every file. sc.textFile(...) will treat all files in given DIR as one file so I think I need to use sc.wholeTextFile which returns below:

res25: Array[(String, String)] =
Array((hdfs://.../bb.txt,"this is bbb1
this is bbb2
this is bbb3
this is bbb4
"), (hdfs://.../aa.txt,"this is aaa1
this is aaa2
this is aaa3
"), (hdfs://.../cc.txt,"this is ccc1
this is ccc2

my code is below but not works, it causes cPickle.PicklingError: Could not serialize object: Py4JError. seems like dataframe.write is not supported in map function

def process(log_lines):
  # after processing log_lines we got dbata
  dbdata= [("1001","nameA","m"),("1002","nameB","f"),("1003","nameC","m")]
  # prepare the data to write to mysql
  schemax = ['id', 'name', 'gender']
  dbdf = spark.createDataFrame(dbdata, schema=schemax)
  dbdf.write.jdbc(url=url, table="mydb", mode='append', properties=prop)

wtf = sc.wholeTextFiles("/home/elogs3/")
rs = wtf.mapValues(lambda txt:txt.split("\n")) .map(lambda j: (process(j[1])))

using rs.collect() then iteration will work, but driver need to use lots of memory Any solution?

0 Answers
Related