Apache Spark-SQL与Sqoop基准测试同时将数据从RDBMS传输到hdfs

Question

我正在研究一个用例,我必须将数据从RDBMS传输到HDFS.我们使用sqoop对此案例进行了基准测试,发现我们能够在6-7分钟内传输大约20GB的数据.

当我尝试使用Spark SQL时,性能非常低(1 GB的记录从netezza转移到hdfs需要4分钟).我正在尝试进行一些调整并提高其性能,但不太可能将其调整到sqoop的水平(1分钟内大约3 Gb的数据).

我同意spark主要是一个处理引擎这一事实,但我的主要问题是spark和sqoop都在内部使用JDBC驱动程序,所以为什么性能上有太大差异(或者可能是我遗漏了一些东西).我在这里发布我的代码.

object helloWorld {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Netezza_Connection").setMaster("local")
    val sc= new SparkContext(conf)
    val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
    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")
    val df2 =sqlContext.sql("select * from POC")
    val partitioner= new org.apache.spark.HashPartitioner(14)
    val rdd=df2.rdd.map(x=>(String.valueOf(x.get(1)),x)).partitionBy(partitioner).values
    rdd.saveAsTextFile("hdfs://Hostname/test")
  }
}

我检查了很多其他帖子,但无法得到sqoop内部工作和调优的明确答案,也没有得到sqoop vs spark sql基准测试.有助于理解这个问题.

Answer 1

您正在使用错误的工具来完成工作.

Sqoop将启动一系列进程(在数据节点上),每个进程都与数据库建立连接(参见num-mapper),它们将分别提取数据集的一部分.我不认为你可以用Spark实现读取并行性.

使用Sqoop获取数据集,然后使用Spark处理它.

Answer 2

您可以尝试以下操作：-

1. 从没有任何分区的 netezza 读取数据，并将 fetch_size 增加到一百万。

   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")
   

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

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

3. ORC 格式比 TEXT 更优化。将最终输出写入 parquet/ORC。

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