SparkSQL: как указать столбец разделения при загрузке набора данных из базы данных

Question

Я использую Spark 2.3 и загружаю данные из MySQL с помощью jdbc, как показано ниже

  val dataSet:Dataset[Row] = _spark
    .read
    .format("jdbc")
    .options(Map("url" -> jdbcUrl
                ,"user" -> username
                ,"password" -> password
                ,"dbtable" -> dataSourceTableName
                ,"driver" -> driver
                ))
    .load() 

Я хотел бы разбить набор данных на основе определенного столбца в таблице.Как мне этого добиться?

Answer 1

spark.read("jdbc")
  .option("url", url)
  .option("dbtable", "pets")
  .option("user", user)
  .option("password", password)
  .option("numPartitions", 10)
  .option("partitionColumn", "owner_id")
  .option("lowerBound", 1)
  .option("upperBound", 10000)

Подробнее по следующей ссылке

• https://medium.com/@radek.strnad/tips-for-using-jdbc-in-apache-spark-sql-396ea7b2e3d3

• http://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

Answer 2

Вам необходимо указать опции partitionColumn, upperBound, lowerBound и numPartitions.

Они описаны в таблице свойств в документации JDBC для spark sql .

Все эти параметры должны быть указаны, если указан какой-либо из них.Кроме того, необходимо указать numPartitions.Они описывают, как разделить таблицу при параллельном чтении от нескольких работников.partitionColumn должен представлять собой числовой столбец, столбец даты или отметки времени из рассматриваемой таблицы.Обратите внимание, что lowerBound и upperBound используются только для определения шага разбиения, а не для фильтрации строк в таблице.Таким образом, все строки в таблице будут разделены и возвращены.Эта опция применяется только к чтению.

Для более подробного объяснения параметров upperBound и lowerBound можно найти @PIYUSH PASARI's answer .

Он приводит следующий пример запросов, сгенерированных со следующими значениями параметров

upperBound = 500, lowerBound = 0 и numPartitions = 5.

SELECT * FROM table WHERE partitionColumn < 100 or partitionColumn is null
SELECT * FROM table WHERE partitionColumn >= 100 AND <200 
SELECT * FROM table WHERE partitionColumn >= 200 AND <300
SELECT * FROM table WHERE partitionColumn >= 300 AND <400
...
SELECT * FROM table WHERE partitionColumn >= 400

Это видно из кода в JDBCRelation.scala .

Как видно, все строки извлекаются, но если ваша верхняя и нижняя границы не покрывают весь диапазон данных, первыйи последние разделы могут быть больше, чем другие.Если вы не можете быть уверены в верхних и нижних границах, хотите иметь четные партиции и не заботитесь о получении каждой строки, вы всегда можете установить верхнюю и нижнюю границы в качестве условий в параметре dbtable.