Как использовать PySpark для потоковой передачи данных в базу данных MySQL? - PullRequest
Новая
       7

Как использовать PySpark для потоковой передачи данных в базу данных MySQL?

0 голосов
/ 12 ноября 2018

В настоящее время я работаю над одностраничным веб-приложением, которое позволяет пользователям загружать большие файлы CSV (в настоящее время тестирует файл размером ~ 7 ГБ) на сервер колб, а затем передавать этот набор данных в базу данных.Загрузка занимает около минуты, и файл полностью сохраняется во временный файл на сервере фляги.Теперь мне нужно иметь возможность передавать этот файл и сохранять его в базе данных.Я провел некоторое исследование и обнаружил, что PySpark отлично подходит для потоковой передачи данных, и я выбираю MySQL в качестве базы данных для потоковой передачи данных CSV (но я открыт для других БД и потоковых методов).Я младший разработчик и новичок в PySpark, поэтому я не уверен, как это сделать.В руководстве по потоковой передаче Spark говорится, что данные должны поступать через такие источники, как Kafka, Flume, сокет TCP и т. Д., Поэтому мне интересно, нужно ли мне использовать какой-либо из этих методов для передачи файла CSV в Spark.Однако я наткнулся на этот замечательный пример , где они передают данные csv в базу данных SQL Azure, и похоже, что они просто читают файл напрямую, используя Spark, без необходимости загружать его через потоковый источник, такой как Kafka, и т. Д.Единственное, что меня смущает в этом примере, - это то, что они используют кластер HDInsight Spark для потоковой передачи данных в базу данных, и я не уверен, как объединить все это с флеш-сервером.Я прошу прощения за отсутствие кода, но в настоящее время у меня есть только файл флеш-сервера с одним маршрутом загрузки файла.Любые примеры, учебные пособия или советы будут с благодарностью.

1 Ответ

0 голосов
/ 13 ноября 2018

Я не уверен насчет потоковой части, но spark может эффективно обрабатывать большие файлы - и сохранение в таблицу db будет выполняться параллельно, поэтому без особых знаний о ваших данных и при условии, что у вас есть загруженный файл на вашем сервере Я бы сказал, что:

Если бы я хотел сохранить большой структурированный файл, такой как csv, в таблице, я бы начал так: 

# 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()

Обратите внимание на режим 'добавление' здесь: главное в том, что spark не может выполнять обновления таблицы, он либо добавляет новые строки, либо заменяет то, что находится в таблице.

Итак, если ваш CSV такой: 

id, name, address....

В результате вы получите таблицу с такими же полями.

Это самый базовый пример, который я мог придумать, чтобы вы начали с искры, без каких-либо соображений относительно искрового кластера или чего-либо еще связанного с этим. Я бы посоветовал вам взять это на спин и выяснить, подходит ли это вашим потребностям :)

Кроме того, имейте в виду, что это может занять несколько секунд или больше в зависимости от ваших данных, места расположения базы данных, вашей машины и загрузки базы данных, поэтому было бы неплохо поддерживать асинхронность ваших API-интерфейсов, опять же, я не знаю ни о каких других ваших деталях.

Надеюсь, это поможет. Удачи!

...