通过Spark SQL批量迁移数据

Question

我目前正在尝试通过Spark SQL将非常大的MySQL表的内容批量迁移到镶木地板文件中.但是当这样做时,即使将驱动程序的内存限制设置得更高(我在本地模式下使用spark),我也会快速耗尽内存.示例代码:

Dataset<Row> ds = spark.read()
    .format("jdbc")
    .option("url", url)
    .option("driver", "com.mysql.jdbc.Driver")
    .option("dbtable", "bigdatatable")
    .option("user", "root")
    .option("password", "foobar")
    .load();

ds.write().mode(SaveMode.Append).parquet("data/bigdatatable");

似乎Spark试图将整个表内容读入内存,这样做不会很好.那么,通过Spark SQL进行批量数据迁移的最佳方法是什么？

Answer 1

在您的解决方案中,Spark会在开始编写之前将整个表内容读入一个分区.您可以避免的一种方法是对读取部分进行分区,但它需要源数据中的数字顺序列:

Dataset<Row> ds = spark.read()
  .format("jdbc")
  .option("url", url)
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "bigdatatable")
  .option("user", "root")
  .option("password", "foobar")
  .option("partitionColumn", "NUMERIC_COL")
  .option("lowerBound", "1")
  .option("upperBound", "10000")
  .option("numPartitions", "64")
  .load();

在上面的例子中,数据中必须存在"NUMERIC_COL"列,理想情况下,它应该从1到10000均匀变化.当然,这是很多要求,而且这样的列可能不存在,所以你应该可能在数据库中使用类似的列创建一个视图,或者在查询中添加它(请注意,我使用了通用SQL语法,您必须适应您的DBMS):

String query = "(select mod(row_number(), 64) as NUMERIC_COL, * from bigdatatable) as foo"

Dataset<Row> ds = spark.read()
  .format("jdbc")
  .option("url", url)
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", query)
  .option("user", "root")
  .option("password", "foobar")
  .option("partitionColumn", "NUMERIC_COL")
  .option("lowerBound", "0")
  .option("upperBound", "63")
  .option("numPartitions", "64")
  .load();