python - 为什么从 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 数据库读取/写入相关的所有最新日志。运行 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