Я пытаюсь переключиться с чтения плоских файлов csv на файлы avro на искру. после https://github.com/databricks/spark-avro Я использую:
import com.databricks.spark.avro._
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val df = sqlContext.read.avro("gs://logs.xyz.com/raw/2016/04/20/div1/div2/2016-04-20-08-28-35.UTC.blah-blah.avro")
и получим
java.lang.UnsupportedOperationException: This mix of union types is not supported (see README): ArrayBuffer(STRING)
в файле readme четко указано:
Эта библиотека поддерживает чтение всех типов Avro, с исключением сложные типы объединения. Он использует следующее сопоставление типов Avro для Типы Spark SQL:
когда я пытаюсь передать текст в тот же файл, я вижу схему
val df = sc.textFile("gs://logs.xyz.com/raw/2016/04/20/div1/div2/2016-04-20-08-28-35.UTC.blah-blah.avro")
df.take(2).foreach(println)
{ "имя" : "log_record", "тип" : "запись", "полей": [{ "имя" : "запрос", "тип" : { "типа": "запись", "имя" : "request_data", "поле": [{ "имя" : "Дата и время", "типа": "строка" }, { "имя" : "ф", "тип" : "строка" }, { "имя" : "хозяин", "тип" : "строка" }, { "имя" : "URI", "тип" : "строка" }, { "имя" : "REQUEST_URI", "тип" : "строка" }, { "имя" : "реферер", "тип" : "строка" }, { "имя" : "UserAgent", "тип" : "строка" }]}}
< ------- отрывок полного ответа ------- >
так как у меня мало контроля над форматом, в который я получаю эти файлы, мой вопрос здесь - есть ли обходной путь, который кто-то тестировал и может рекомендовать?
Я использую gc dataproc с
MASTER = прямая кластерная искровая оболочка -num-executors 4 --executor-memory 4G -executor-core 4 --packages com.databricks: spark-avro_2.10: 2.0.1, com.databricks: искровой csv_2.11:1.3.0
любая помощь будет принята с благодарностью.....
Вы найдете любое решение, которое работает с Spark SQL. Каждый столбец в Spark должен содержать значения, которые могут быть представлены как один DataType, поэтому сложные типы объединения просто не представляются с помощью Spark Dataframe,
Если вы хотите читать такие данные, вы должны использовать RDD API и позже конвертировать загруженные данные в Dataframe.