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

Question

最终，我想将数据从 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)

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

Answer 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