为啥从 Spark 写入 Vertica DB 比从 Spark 写入 MySQL 需要更长的时间？

Posted 2023-03-27

【中文标题】为啥从 Spark 写入 Vertica DB 比从 Spark 写入 MySQL 需要更长的时间？

【英文标题】：Why does writing from Spark to Vertica DB take longer than writing from Spark to MySQL?为什么从 Spark 写入 Vertica DB 比从 Spark 写入 MySQL 需要更长的时间？

【发布时间】：2016-04-19 17:39:44

【问题描述】：

最终，我想将数据从 Vertica DB 抓取到 Spark，训练机器学习模型，进行预测，然后将这些预测存储到另一个 Vertica DB。

当前问题是确定流程最后一部分的瓶颈：将值从 Spark 存储在 Vertica DB 中。在 Vertica DB 中存储 63k 行数据大约需要 38 分钟。相比之下，当我将相同的数据从 Spark 传输到 MySQL 数据库 时，需要 10 秒。

我不知道为什么差异如此之大。

我有分别用于 Vertica 和 mysql 连接的名为 VerticaContext 和 MySQLContext 的类。这两个类都使用SQLContext 来读取使用jdbc 格式的条目。

df = self._sqlContext.read.format('jdbc').options(url=self._jdbcURL, dbtable=subquery).load()

并使用 jdbc 编写。

df.write.jdbc(self._jdbcURL, table_name, save_mode)

除了写入不同的目标数据库之外，这两个类之间没有区别。我很困惑为什么保存表格所需的时间有很大差异。是不是因为两个不同数据库在硬件上的内在差异？

【问题讨论】：

可能是因为 Vertica 是列式数据库？

Spark 是在做单次 INSERT 吗？

您的问题中的详细信息太少。您的 Vertica 数据库可能未针对您通过 JDBC 执行的涓流插入进行优化。您需要与您的 DBA 交谈。

在应用程序端，您可以尝试流式插入，如 here 所述

@mustaccio，感谢您的链接。不知道有没有和这个等价的python？

【参考方案1】：

我想出了一个替代解决方案。在深入研究之前，我将解释我的发现以及为什么我认为保存到 Vertica DB 的速度很慢。

Vertica 日志（在您的 Vertica 计算机上搜索文件“vertica.log”）包含与 Vertica DB 的读取/写入相关的所有最新日志。运行 write 命令后，我发现这实际上是在 Vertica DB 中创建 INSERT 语句。 INSERT 语句（没有“DIRECT”指令）很慢，因为它们被写入 WOS（RAM）而不是 ROS（磁盘）。我不知道为什么会这样的确切细节。写入正在发出单独的 INSERT 语句 插入速度慢是一个已知问题。我很难找到这些信息，但我终于找到了一些支持该信息的链接。我将它们放在这里以供后代使用：http://www.vertica-forums.com/viewtopic.php?t=267, http://vertica-forums.com/viewtopic.php?t=124

我的解决方案：

documentation 表示 COPY 命令（带有“DIRECT”关键字）是将大量数据加载到数据库的最有效方式。由于我一直在寻找 Python 解决方案，因此我使用了 Uber 的 vertica-python package，它允许我与 Vertica DB 建立连接并发送 Vertica 命令来执行。

我想利用 COPY 命令的效率，但数据位于 Vertica 集群之外的某个位置。我需要将数据从我的 Spark 集群发送到 Vertica DB。幸运的是，HDFS 有一种方法可以做到这一点（请参阅here）。我决定将数据帧转换为 csv 文件并将其保存在 HDFS 上。然后我将 COPY 命令发送到 Vertica DB 以从 HDFS 获取文件。

我的代码如下（假设我已经有一个存储 pyspark 数据帧的变量。我们称之为“df”）：

    import vertica_python as VertPy

    df.toPandas().to_csv(hdfs_table_absolute_filepath, header=False, index=False)
    conn_info = 
        'host': ['your-host-here']
        'port': [port #],
        'user': ['username'],
        'password': ['password'],
        'database': ['database']
    

    conn = VertPy.connect(**conn_info)
    cur = conn.cursor()

    copy_command = create_copy_command(table_name, hdfs_table_relative_filepath)
    cursor.execute(copy_command)

    def create_copy_command(table_name, table_filepath):
        copy_command = "COPY " + table_name + " SOURCE Hdfs(url='http://hadoop:50070/webhdfs/v1" + table_filepath + "', username='root') DELIMITER ',' DIRECT ABORT ON ERROR"
        return copy_command

【讨论】：

问题是它正在执行多个insert...values，这在 Vertica 中非常低效。 DIRECT 如果您在短时间内对同一张表执行大量事务，则需要小心处理，因为每个事务都会创建一个 ROS 容器。不过，使用 COPY 是正确的。此外，如果您的数据在您的数据框中，您可以使用vertica_python 的复制方法并以这种方式流式传输数据，而不是使用SOURCE...您将使用FROM STDIN。