问题标签 [spark-jdbc]

• 火花 2.2.0

我从 SQL 脚本转换了以下代码。它已经运行了两个小时，它仍在运行。甚至比 SQL Server 还要慢。有什么事情做得不对吗？

以下是计划，

1. 推table2送给所有执行者
2. 分区table1并将分区分发给执行者。
3. 并且每一行table2/t2连接（交叉连接）的每个分区table1。

因此交叉连接结果的计算可以分布式/并行运行。（我想，例如假设我有 16 个 executor，在所有 16 个 executor 上保留一份 t2 的副本。然后将表 1 分成 16 个分区，每个 executor 一个。然后每个 executor 对表 1 的一个分区进行计算和 t2。）

如何用 Spark 惯用的方式重写它？

参考：https ://forums.databricks.com/questions/6747/how-do-i-get-a-cartesian-product-of-a-huge-dataset.html

你知道以下行是否可以处理 jdbc 池连接：

你有什么主意吗？谢谢

我尝试过使用 statement.setFetchSize(required number)，但是当我使用 JDBC 连接到 Hive 时，它​​有效，而不是当我尝试通过 spark thrift 服务器时。我的查询产生了一个大的结果集，导致在 thrift 服务器上出现 OOM。

有什么设置可以用来做同样的事情吗？

我正在尝试使用 SparkSQL JDBC Writer 将 hive 表中的 5000 万行插入到 SQLServer 表中。下面是我用来插入数据的代码行

mdf1.coalesce(4).write.mode(SaveMode.Append).jdbc(connectionString, "dbo.TEST_TABLE", connectionProperties)

处理 1000 万行并出现以下错误后，火花作业失败

java.sql.BatchUpdateException：SQL Server 数据库引擎的实例此时无法获取 LOCK 资源。当活动用户较少时重新运行您的语句。要求数据库管理员检查此实例的锁和内存配置，或检查长时间运行的事务。

但是如果我使用下面的代码行，同样的工作也会成功。

mdf1.coalesce(1).write.mode(SaveMode.Append).jdbc(connectionString, "dbo.TEST_TABLE", connectionProperties)

我正在尝试打开 4 个与 SQLServer 的并行连接以优化性能。但是在处理了 1000 万行之后，该作业不断失败，并出现“无法获取锁错误”。此外，如果我将数据帧限制为几百万行（少于 1000 万行），即使有四个并行连接，作业也会成功

任何人都可以建议我是否可以使用 SparkSQL 将大量数据导出到 RDBMS 中，以及我是否需要对 SQL 服务器表进行任何配置更改。

提前致谢。

我正在使用查询从 MYSQL 中获取数据，如下所示：

问题是，我可以在我指定的位置使用伪列（如ROWNUM在 Oracle 或RRN(employeeno)DB2 中）吗？optionpartitionColumn

如果不是，我们可以指定一个不是主键的分区列吗？

我有一个包含数百万条记录的巨大数据库表。每条记录都可以单独处理，并且必须转换为，比如说，一个字符串。

所以我开始环顾四周，我想知道 Spark 在这种情况下是否可以帮助我。具体来说，我写了一些非常简单的东西：

问题：它适用于小型/中型表，但在大型表的情况下我会得到 OutOfMemory。

即使我认为我知道 jdbc 分区是如何工作的（并且它正在工作），似乎 session.read.jdbc 只有在整个数据集加载后才返回（即将每一行移动到 map 方法）。

是否有可能使用这种或另一种方法在读取每一行时转换（即处理）每一行？

我已经看过下面评论中指出的类似问题，但是在那里，那个用户正在做一个聚合（df.count），而我只需要一个一个地迭代记录，所以我想知道这是否“懒惰”迭代是可能的。

谢谢

我一直在研究基于 pyspark 的内容拉/同步工具（类似于 sqoop，但应用了一些转换作为管道）。我正在通过以下方式创建 jdbc 连接：

(...)

在 pyspark repl 中，我可以确认我有 4 个分区，但是我不知道如何判断每个分区的上/下份额是多少。

当我运行保存时，我最终会很快完成 3 个执行程序/任务，最后一个执行程序完成所有工作。当我扩展到一个非常大的（10 亿行 +）表时，同样的事情 - 200 个任务，其中 199 个在 < 100 毫秒内完成，最后一个完成所有工作。

我有几个问题：

1. 如何调试每个分区/任务集是什么
2. 如何查看每个任务的查询是什么（类似于 #1 ）

我尝试将查询转换为 spark.sql 并创建临时视图等。那里的优化器做同样的事情。（有意义的是它共享相同的数据框/连接设置）

上游是 mssql jdbc 驱动程序 mssql-jdbc-6.2.2.jre8 ， jdk 1.8.112 ( oracle )。pyspark 2.2.0.2.6.3.0-235 (hdp 2.6.3)

我对numPartitions以下方法中参数的行为感到困惑：

1. DataFrameReader.jdbc
2. Dataset.repartition

关于参数的官方文档如下DataFrameReader.jdbcnumPartitions

numPartitions：分区数。这与 lowerBound（包括）、upperBound（不包括）一起形成了用于生成的 WHERE 子句表达式的分区步长，这些表达式用于均匀地拆分列 columnName。

和官方文档Dataset.repartition说

返回具有完全numPartitions分区的新数据集。

我目前的理解：

1. 方法中的numPartition参数DataFrameReader.jdbc控制从数据库中读取数据的并行度
2. 中的numPartition参数Dataset.repartition控制将其写入磁盘时将生成的输出文件的数量DataFrame

我的问题：

1. 如果我DataFrame通过读取DataFrameReader.jdbc然后将其写入磁盘（不调用repartition方法），那么输出中的文件是否仍然与我DataFrame在调用磁盘后将其写入磁盘一样多repartition？
2. 如果上述问题的答案是：
   • 是的：那么在使用方法（带参数）读取repartition的方法上调用方法是多余的吗？DataFrameDataFrameReader.jdbcnumPartitions
   • 否：那么请纠正我理解的失误。同样在这种情况下，方法的numPartitions参数不应该DataFrameReader.jdbc被称为“并行”之类的东西吗？

我是 Spark 的新手，我们目前正在使用 spark-java 从 Oracle 数据库创建 orc 文件。我能够配置连接

但是，我在属性中找不到任何方法来指定 trustStore 或 trustStoreType。有人可以帮助我了解如何指定这些属性吗？

我已经尝试将属性填充为

但这对我不起作用

更新 1：我已经尝试了 user8371915 的建议，并将 sso 文件放在我的两个执行程序节点中。我仍然收到以下异常（精简版）

是否可以将参数注入到 spark JDBC 插入语句中？

我在用着

用于将批量数据帧保存到我的数据库。

在JdbcUtils被insertStatement创建。

是否有可能（不创建我自己的 JDBC 连接和语句）在语句中添加“错误中止”之类的参数？