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?