Столбец приведения кадра данных Spark для совместимости с Kudu

(Я новичок в Spark, Impala и Kudu.) Я пытаюсь скопировать таблицу из базы данных Oracle в таблицу Impala с той же структурой в Spark через Kudu. Я получаю сообщение об ошибке, когда код пытается сопоставить Oracle NUMBER с типом данных Kudu. Как изменить тип данных Spark DataFrame, чтобы сделать его совместимым с Kudu?

Предполагается, что это будет копия данных 1-к-1 из Oracle в Impala. Я извлек схему Oracle из исходной таблицы и создал целевую таблицу Impala с той же структурой (те же имена столбцов и разумное сопоставление типов данных). Я надеялся, что Spark+Kudu автоматически сопоставит все и просто скопирует данные. Вместо этого Kudu жалуется, что не может сопоставить DecimalType(38,0).

Я хотел бы указать, что «столбец № 1 с именем SOME_COL, который является NUMBER в Oracle, должен быть сопоставлен с LongType, который поддерживается в Kudu».

Как это сделать?

// This works
val df: DataFrame = spark.read
  .option("fetchsize", 10000)
  .option("driver", "oracle.jdbc.driver.OracleDriver")
  .jdbc("jdbc:oracle:thin:@(DESCRIPTION=...)", "SCHEMA.TABLE_NAME", partitions, props)

// This does not work  
kuduContext.insertRows(df.toDF(colNamesLower: _*), "impala::schema.table_name")
// Error: No support for Spark SQL type DecimalType(38,0)
// See https://github.com/cloudera/kudu/blob/master/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/SparkUtil.scala

// So let's see the Spark data types
df.dtypes.foreach{case (colName, colType) => println(s"$colName: $colType")}
// Spark  data type: SOME_COL DecimalType(38,0)
// Oracle data type: SOME_COL NUMBER -- no precision specifier; values are int/long
// Kudu   data type: SOME_COL BIGINT

apache-spark scala impala apache-kudu
person radumanolescu    schedule 15.05.2019    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Очевидно, мы можем указать пользовательскую схему при чтении из источника данных JDBC.

connectionProperties.put("customSchema", "id DECIMAL(38, 0), name STRING")
val jdbcDF3 = spark.read
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

Это сработало. Я смог указать customSchema так:

col1 Long, col2 Timestamp, col3 Double, col4 String

и с этим код работает:

import spark.implicits._
val df: Dataset[case_class_for_table] = spark.read
  .option("fetchsize", 10000)
  .option("driver", "oracle.jdbc.driver.OracleDriver")
  .jdbc("jdbc:oracle:thin:@(DESCRIPTION=...)", "SCHEMA.TABLE_NAME", partitions, props)
  .as[case_class_for_table]
kuduContext.insertRows(df.toDF(colNamesLower: _*), "impala::schema.table_name")
person radumanolescu    schedule 16.05.2019