如何使用 PySpark 更新 hive 表中的记录？

Question

我们正在使用 spark 处理大数据，最近有了新的用例，我们需要使用 spark 更新 Hive 表中的数据。

下面是一个简单的例子： 数据驻留在 Hive 表中，应用程序使用 PySpark 读入数据帧（例如 df1）。 例如：数据框有下面的列。

EmpNo Name 年龄薪水

1 aaaa 28 30000

2 bbbb 38 20000

3 cccc 26 25000

4 dddd 30 32000

需要使用 spark 向表中添加更多记录。

例如：

Action EmpNo Name 年龄薪水

添加 5 dddd 30 32000

应用程序可以通过剥离 Action 列并将新数据附加到表中，将新数据读入第二个数据帧（例如 df2）。它很简单，而且效果很好。

df.write.format('镶木地板') \ .mode('追加') \ .saveAsTable(canonical_hive_table)

在某些情况下，我们需要删除现有记录或根据 Action 列更新它们。

例如：

Action EmpNo Name 年龄薪水

删除 2 bbbb 38 20000

更新 4 dddd 30 42000

在上面的例子中，应用程序需要删除 EmpNo:2 并更新 EmpNo:4。

最终输出应如下所示：

EmpNo Name 年龄薪水

1 aaaa 28 30000

3 cccc 26 25000

4 dddd 30 42000

5 dddd 30 32000

据我了解，更新操作在 Spark Sql 中不可用，而且数据框是不可变的，无法更改记录。

有人遇到过这种情况吗？或知道使用 PySpark 更新 Hive 表中现有记录的任何选项？

请：应用程序需要定期处理数百万条记录的数千条更新。

提前致谢。

Answer 1

在大多数情况下，您需要使用正确的工具和方法 - 并意识到任何限制； Hadoop 大体上是不可变的。

Data Frames 可以以不同的模式保存，但不能选择性地更新记录 -pyspark 或 scala 或 R 或​​ JAVA 与它无关，除了 Cloudera 的 KUDU 存储管理器，它有一个 DF Writer 可以使用的 Spark 连接器，但是当我上次使用它时，管理员不喜欢安全限制。

你能做什么？

1. 使用可变的 KUDU - 但我怀疑不是一个选项。它就像 PARQUET - 柱状，其性能与我的经验相当。
2. 是否使用您的 DF，并使用 ORC 文件而不是 PARQUET 来实现持久性，这些文件也是列式的，可以使用脚本中的 Hive 的 MERGE 语句或启用 Hive 支持的 Spark SQL 进行更新。此选项意味着忘记 PARQUET。此链接提供了一些有趣的见解：https://www.quora.com/What-are-the-differences-between-ORC-Avro-and-Parquet-File-Formats-in-Hadoop-in-terms-of-compression-and-speed
3. 在 DF 中做你的事情，什么不做，重新声明（即再次写入）所有数据并写入 PARQUET 表/目录的 2 个版本之一，无论是否分区，以及添加视图层以在当前和新版本视图之间切换。这是在不使用 ORC 时完成的。

使用 MERGE，可以压缩分区内的小型 Hadoop Hive 文件，但前提是格式为 ORC - 如果内存服务正确。我将不得不在这里刷新我的记忆，可能是它在新版本和 API 中发生了变化。

此外：

1. 因此，您有几种选择，无需在 SPARK 中使用 DF 来完成所有操作。
2. Databricks 也有此类数据的 DELTA 表。
3. 如果您使用来自 RDBMS.s 的 JDBC 导入，您可以以 akward 方式将 sqoop 与外部表一起使用以获取更新的数据，但我无法从问题中看出这一点。这是逻辑：sqoop merge-key creating multiple part files instead of one which doesn't serve the purpose of using merge-key

需要考虑的一些事情。祝你好运。