java - 与 csv 文件相比，将 mysql 表转换为 spark 数据集非常慢

Question

我在 Amazon s3 中有一个大小为 62mb(114000 行)的 csv 文件。我正在将它转换为 spark 数据集，并从中获取前 500 行。代码如下；

DataFrameReader df = new DataFrameReader(spark).format("csv").option("header", true);
Dataset<Row> set=df.load("s3n://"+this.accessId.replace("\"", "")+":"+this.accessToken.replace("\"", "")+"@"+this.bucketName.replace("\"", "")+"/"+this.filePath.replace("\"", "")+"");

 set.take(500)

整个操作需要 20 到 30 秒。

现在我正在尝试相同的方法，但我使用的是 csv，我使用的是具有 119 000 行的 mySQL 表。 MySQL 服务器在 amazon ec2 中。代码如下；

String url ="jdbc:mysql://"+this.hostName+":3306/"+this.dataBaseName+"?user="+this.userName+"&password="+this.password;

SparkSession spark=StartSpark.getSparkSession();

SQLContext sc = spark.sqlContext();

DataFrameReader df = new DataFrameReader(spark).format("csv").option("header", true);
Dataset<Row> set = sc
            .read()
            .option("url", url)
            .option("dbtable", this.tableName)
            .option("driver","com.mysql.jdbc.Driver")
            .format("jdbc")
            .load();
set.take(500);

这需要 5 到 10 分钟。我在 jvm 中运行 spark。在这两种情况下使用相同的配置。

我可以使用 partitionColumn、numParttition 等，但我没有任何数字列，还有一个问题是我不知道表的架构。

我的问题不是如何减少所需的时间，因为我知道在理想情况下 spark 将在集群中运行，但我不明白的是为什么上述两种情况下的时间差异如此之大？

Answer 1

此问题已在 StackOverflow 上多次提及:

• How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
• spark jdbc df limit... what is it doing?
• How to use JDBC source to write and read data in (Py)Spark?

在外部资源中:

• https://github.com/awesome-spark/spark-gotchas/blob/master/05_spark_sql_and_dataset_api.md#parallelizing-reads

所以重申一下 - 默认情况下 DataFrameReader.jdbc 不分发数据或读取。它使用单线程、单执行器。

分发阅读:

• 使用 lowerBound/upperBound 的范围:

  Properties properties;
  Lower
  
  Dataset<Row> set = sc
      .read()
      .option("partitionColumn", "foo")
      .option("numPartitions", "3")
      .option("lowerBound", 0)
      .option("upperBound", 30)
      .option("url", url)
      .option("dbtable", this.tableName)
      .option("driver","com.mysql.jdbc.Driver")
      .format("jdbc")
      .load();
  

• 谓词

  Properties properties;
  Dataset<Row> set = sc
      .read()
      .jdbc(
          url, this.tableName,
          {"foo < 10", "foo BETWWEN 10 and 20", "foo > 20"},
          properties
      )