Чтение CSV в Spark Dataframe с отметками времени и типами даты

Это CDH с Spark 1.6.

Я пытаюсь импортировать этот Гипотетический CSV в apache Spark DataFrame:

$ hadoop fs -cat test.csv
a,b,c,2016-09-09,a,2016-11-11 09:09:09.0,a
a,b,c,2016-09-10,a,2016-11-11 09:09:10.0,a

Я использую databricks-csv jar.

val textData = sqlContext.read
    .format("com.databricks.spark.csv")
    .option("header", "false")
    .option("delimiter", ",")
    .option("dateFormat", "yyyy-MM-dd HH:mm:ss")
    .option("inferSchema", "true")
    .option("nullValue", "null")
    .load("test.csv")

Я использую inferSchema для создания схемы для результирующего DataFrame. Функция printSchema() дает мне следующий вывод для кода выше:

scala> textData.printSchema()
root
 |-- C0: string (nullable = true)
 |-- C1: string (nullable = true)
 |-- C2: string (nullable = true)
 |-- C3: string (nullable = true)
 |-- C4: string (nullable = true)
 |-- C5: timestamp (nullable = true)
 |-- C6: string (nullable = true)

scala> textData.show()
+---+---+---+----------+---+--------------------+---+
| C0| C1| C2|        C3| C4|                  C5| C6|
+---+---+---+----------+---+--------------------+---+
|  a|  b|  c|2016-09-09|  a|2016-11-11 09:09:...|  a|
|  a|  b|  c|2016-09-10|  a|2016-11-11 09:09:...|  a|
+---+---+---+----------+---+--------------------+---+

Столбец C3 имеет тип String. Я хочу, чтобы C3 имел тип date. Чтобы получить его до даты, я попробовал следующий код.

val textData = sqlContext.read.format("com.databricks.spark.csv")
    .option("header", "false")
    .option("delimiter", ",")
    .option("dateFormat", "yyyy-MM-dd")
    .option("inferSchema", "true")
    .option("nullValue", "null")
    .load("test.csv")

scala> textData.printSchema
root
 |-- C0: string (nullable = true)
 |-- C1: string (nullable = true)
 |-- C2: string (nullable = true)
 |-- C3: timestamp (nullable = true)
 |-- C4: string (nullable = true)
 |-- C5: timestamp (nullable = true)
 |-- C6: string (nullable = true)

scala> textData.show()
+---+---+---+--------------------+---+--------------------+---+
| C0| C1| C2|                  C3| C4|                  C5| C6|
+---+---+---+--------------------+---+--------------------+---+
|  a|  b|  c|2016-09-09 00:00:...|  a|2016-11-11 00:00:...|  a|
|  a|  b|  c|2016-09-10 00:00:...|  a|2016-11-11 00:00:...|  a|
+---+---+---+--------------------+---+--------------------+---+

Единственная разница между этим кодом и первым блоком - это опция dateFormat (я использую "yyyy-MM-dd" вместо "yyyy- MM-dd HH: mm: ss" ). Теперь я получаю как C3, так и C5 как timestamps (C3 еще не является датой). Но для C5 часть HH:: mm: ss игнорируется и отображается как нули в данных.

В идеале я хочу, чтобы C3 имел дату типа, C5 - временную метку типа и ее HH: mm: ss часть, которую нельзя игнорировать. Мое решение прямо сейчас выглядит так. Я делаю csv, вытягивая данные параллельно с моей БД. Я уверен, что я вытаскиваю все даты как временные метки (не идеально). Итак, тестовый csv выглядит следующим образом:

$ hadoop fs -cat new-test.csv
a,b,c,2016-09-09 00:00:00,a,2016-11-11 09:09:09.0,a
a,b,c,2016-09-10 00:00:00,a,2016-11-11 09:09:10.0,a

Это мой последний рабочий код:

val textData = sqlContext.read.format("com.databricks.spark.csv")
    .option("header", "false")
    .option("delimiter", ",")
    .option("dateFormat", "yyyy-MM-dd HH:mm:ss")
    .schema(finalSchema)
    .option("nullValue", "null")
    .load("new-test.csv")

Здесь я использую полный формат timestamp ("yyyy-MM-dd HH: mm: ss" ) в dateFormat. Я вручную создаю экземпляр finalSchema, где c3 - дата, а C5 - тип Timestamp (Spark sql types). Я применяю эту схему, используя функцию schema(). Результат выглядит следующим образом:

scala> finalSchema
res4: org.apache.spark.sql.types.StructType = StructType(StructField(C0,StringType,true), StructField(C1,StringType,true), StructField(C2,StringType,true), StructField(C3,DateType,true), StructField(C4,StringType,true), StructField(C5,TimestampType,true), StructField(C6,StringType,true))

scala> textData.printSchema()
root
 |-- C0: string (nullable = true)
 |-- C1: string (nullable = true)
 |-- C2: string (nullable = true)
 |-- C3: date (nullable = true)
 |-- C4: string (nullable = true)
 |-- C5: timestamp (nullable = true)
 |-- C6: string (nullable = true)


scala> textData.show()
+---+---+---+----------+---+--------------------+---+
| C0| C1| C2|        C3| C4|                  C5| C6|
+---+---+---+----------+---+--------------------+---+
|  a|  b|  c|2016-09-09|  a|2016-11-11 09:09:...|  a|
|  a|  b|  c|2016-09-10|  a|2016-11-11 09:09:...|  a|
+---+---+---+----------+---+--------------------+---+

Есть ли более простой или изящный способ разбора csv файла (который имеет как дату, так и временную метку в блок данных искры?

Релевантные ссылки:
http://spark.apache.org/docs/latest/sql-programming-guide.html#manually-specifying-options
https://github.com/databricks/spark-csv

Ответ 1

С опцией infer для нетривиальных случаев она, вероятно, не вернет ожидаемый результат. Как вы можете видеть в InferSchema.scala:

if (field == null || field.isEmpty || field == nullValue) {
  typeSoFar
} else {
  typeSoFar match {
    case NullType => tryParseInteger(field)
    case IntegerType => tryParseInteger(field)
    case LongType => tryParseLong(field)
    case DoubleType => tryParseDouble(field)
    case TimestampType => tryParseTimestamp(field)
    case BooleanType => tryParseBoolean(field)
    case StringType => StringType
    case other: DataType =>
      throw new UnsupportedOperationException(s"Unexpected data type $other")

Он будет пытаться сопоставить каждый столбец с типом метки, а не с типом даты, поэтому "из коробки решение" для этого случая невозможно. Но, по моему опыту, "более легкое" решение напрямую определяет схему с нужный тип, это позволит избежать опции infer, задающей тип, который только совпадения для RDD оценивали не все данные. Ваша окончательная схема - эффективное решение.

Ответ 2

Это не очень элегантно, но вы можете преобразовать метку времени в дату следующим образом (проверьте последнюю строку):

val textData = sqlContext.read.format("com.databricks.spark.csv")
    .option("header", "false")
    .option("delimiter", ",")
    .option("dateFormat", "yyyy-MM-dd")
    .option("inferSchema", "true")
    .option("nullValue", "null")
    .load("test.csv")
    .withColumn("C4", expr("""to_date(C4)"""))