CODEWITHSUNDEEP
pythonmysqlflaskpysparkspark-streaming
Mario
Mario

Reputation: 19

How to use PySpark to stream data into MySQL database?

I am currently working on a single page web app that allows users to upload large CSV files (currently testing a ~7GB file) to a flask server and then stream that dataset to a database. The upload takes about a minute and the file gets fully saved to a temporary file on the flask server. Now I need to be able to stream this file and store it into a database. I did some research and found that PySpark is great for streaming data and I am choosing MySQL as the database to stream the CSV data into (but I am open to other dbs and streaming methods). I am a junior dev and new to PySpark so I'm not sure how to go about this. The Spark streaming guide says that data must ingested through a source like Kafka, Flume, TCP socets, etc so I am wondering if I have to use any of those methods to get my CSV file into Spark. However, I came across this great example where they are streaming csv data into Azure SQL database and it looks like they are just reading the file directly using Spark without needing to ingest it through a streaming source like Kafka, etc. The only thing that confuses me with that example is that they are using a HDInsight Spark cluster to stream data into the db and I'm not sure how to incorporate that all with a flask server. I appologize for the lack of code but currently I just have a flask server file with one route doing the file upload. Any examples, tutorials, or advice would be greatly appreciated.

Upvotes: 1

Views: 2667

Answers (1)

mkaran
mkaran

Reputation: 2718

I am not sure about the streaming part but spark can handle large files efficiently - and storing to a db table will be done in parallel, so without much knowledge about your details, and provided that you have the uploaded file on your server, I'd say that:

If I wanted to save a big structured file like a csv in a table, I would start like this:

# start with some basic spark configuration, e.g. we want the timezone to be UTC 
conf = SparkConf()
conf.set('spark.sql.session.timeZone', 'UTC')
# this is important: you need to have the mysql connector jar for the right mysql version:
conf.set('jars', 'path to mysql connector jar you can download from here: https://dev.mysql.com/downloads/connector/odbc/')
# instantiate a spark session: the first time it will take a few seconds
spark = SparkSession.builder \
    .config(conf=conf) \
    .appName('Huge File uploader') \
    .getOrCreate()

# read the file first as a dataframe
df = spark.read.csv('path to 7GB/ huge csv file')

# optionally, add a filename column
from pyspark.sql import functions as F
df = df.withColumn('filename', F.lit('thecurrentfilename'))

# write it to the table
df.write.format('jdbc').options(
            url='e.g. localhost:port',
            driver='com.mysql.cj.jdbc.Driver',  # the driver for MySQL
            dbtable='the table name to save to',
            user='user',
            password='secret',
        ).mode('append').save()

Note the mode 'append' here: the catch in this is that spark cannot perform updates on a table, it is either append the new rows or replace what is in the table.

So, if your csv is like this:

id, name, address....

You will end up with a table with the same fields.

This is the most basic example I could think of so that you start with spark, with no considerations about a spark cluster or anything else related. I would suggest you take this for a spin and figure out if this suits your needs :)

Also, keep in mind that this might take a few seconds or more depending on your data, where the database is located, your machine and your database load, so it might be a good idea to keep things asynchronous with your api, again I don't know about any of your other details.

Hope this helps. Good luck!

Upvotes: 1

Related Questions