sqoop和sparksql都使用jdbc连接从rdbms引擎获取数据，但是sqoop在这里有一个优势，因为它专门用于在rdbms和hdfs之间迁移数据。
sqoop中可用的每个选项都经过了微调，以在执行数据摄取时获得最佳性能。
您可以从讨论控制Map器数量的选项-m开始。
这就是从rdbms并行获取数据所需要做的。我能用sparksql做吗？当然可以，但是开发人员需要处理sqoop自动处理的“多线程”。

你用错工具了。
sqoop将启动一系列进程（在datanodes上），每个进程都将连接到您的数据库（请参阅num mapper），并且每个进程都将提取数据集的一部分。我不认为你能用spark实现读并行。
使用sqoop获取数据集，然后使用spark进行处理。

下面的解决方案帮助了我

var df=spark.read.format("jdbc").option("url","
"url").option("user","user").option("password","password").option("dbTable","dbTable").option("fetchSize","10000").load()
df.registerTempTable("tempTable")
var dfRepart=spark.sql("select * from tempTable distribute by primary_key") //this will repartition the data evenly

dfRepart.write.format("parquet").save("hdfs_location")

我也遇到了同样的问题，因为您正在使用的代码不适用于分区。

sqlContext.read.format("jdbc").option("url","jdbc:netezza://hostname:port/dbname").option("dbtable","POC_TEST").option("user","user").option("password","password").option("driver","org.netezza.Driver").option("numPartitions","14").option("lowerBound","0").option("upperBound","13").option("partitionColumn", "id").option("fetchSize","100000").load().registerTempTable("POC")

您可以通过以下方式检查spark作业中创建的分区数

df.rdd.partitions.length

可以使用以下代码连接数据库：

sqlContext.read.jdbc(url=db_url,
    table=tableName,
    columnName="ID",
    lowerBound=1L,
    upperBound=100000L,
    numPartitions=numPartitions,
    connectionProperties=connectionProperties)

为了优化你的spark工作，以下是参数：1.#分区2--num executors 3。--executor核心4--执行器存储器5--驾驶员记忆6。获取大小
2、3、4和5选项取决于集群配置，您可以在spark ui上监视spark作业。

@阿弥陀佛虽然是一个答案，但我不同意。
一旦您在从jdbc读取数据时给出了对数据进行分区的 predicate ，spark将为每个分区运行单独的任务。在您的情况下，任务数应该是14（您可以使用spark ui确认这一点）。
我注意到您使用local作为master，这将只为执行者提供1个核心。因此不会有平行性。你的案子就是这样。
现在要获得与sqoop相同的吞吐量，您需要确保这些任务是并行运行的。理论上，这可以通过以下两种方法来实现：1。使用14个执行器，每个执行器有1个核心2个。使用1个14核的执行器（频谱的另一端）
通常，每个执行器有4-5个内核。所以我用15/5=3个执行器来测试性能（我添加了1到14个执行器来考虑在clustor模式下运行的驱动程序的一个内核）。使用：executor.cores，executor.instances in sparkconf.set来播放配置。
如果这不能显著提高性能，那么接下来的事情就是查看执行器内存。
最后，我将调整应用程序逻辑以查看maprdd大小、分区大小和无序排列大小。

你可以试试这个following:-
从netezza中读取数据，不需要任何分区，并且大小增加到一百万。

sqlContext.read.format("jdbc").option("url","jdbc:netezza://hostname:port/dbname").option("dbtable","POC_TEST").option("user","user").option("password","password").option("driver","org.netezza.Driver").option("fetchSize","1000000").load().registerTempTable("POC")

在将数据写入最终文件之前重新划分数据。

val df3 = df2.repartition(10) //to reduce the shuffle

orc格式比文本更优化。将最终输出写入parquet/orc。

df3.write.format("ORC").save("hdfs://Hostname/test")