Приложение Spark с Hive Warehouse Connector неправильно сохраняет поля массива и карты в таблице Hive - PullRequest
Новая
       84

Приложение Spark с Hive Warehouse Connector неправильно сохраняет поля массива и карты в таблице Hive

2 голосов
/ 01 апреля 2020

Я использую Spark 2.4 с Hive Warehouse Connector и Scala 2.11. Текущий Hive Warehouse Connector, предоставленный Hortonworks, не совместим с Spark 2.4. Поэтому я компилирую свой jar-файл из https://github.com/abh1sh2k/spark-llap/pull/1/files, что делает его совместимым со Spark 2.4.

Приложение My Spark читает из входного потока Kafka и записывает в таблицу Hive (формат OR C) используя выходной поток Hive, предоставленный Hive Warehouse Connector.

Вот мой код Spark (Scala): 

package example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col
import za.co.absa.abris.avro.read.confluent.SchemaManager
import za.co.absa.abris.avro.functions.from_confluent_avro

object NormalizedEventsToHive extends Logging {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("NormalizedEventsToHive")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val schema_registry_config = Map(
      "schema.registry.url"           -> "http://schema-registry:8081",
      "value.schema.naming.strategy"  -> "topic.name",
      "schema.registry.topic"         -> "events-v1",
      "value.schema.id"               -> "latest"
    )

    val input_stream_df = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "kafka:9092")
      .option("startingOffsets", "earliest")
      .option("subscribe", "events-v1")
      .load()

    val data = input_stream_df
      .select(from_confluent_avro(col("value"), schema_registry_config) as 'data)
      .select("data.*")

    val output_stream_df = data.writeStream.format("com.hortonworks.spark.sql.hive.llap.streaming.HiveStreamingDataSource")
      .option("database", "default")
      .option("table", "events")
      .option("checkpointLocation", "file:///checkpoint2")
      .option("metastoreUri", "thrift://hive-metastore:9083")
      .start()

    output_stream_df.awaitTermination()
  }
}

Входные данные Kafka messages - это AVRO encoded и используется Confluent Schema Registry для контроля версий схемы. za.co.absa.abris.avro.functions.from_confluent_avro используется для декодирования сообщений Кафки в кодировке AVRO.

Вот схема AVRO: 

{
    "type": "record",
    "name": "events",
    "fields": [
        { "name": "id",                 "type": ["null", "string"], "default": null },
        .....
        { "name": "field_map",          "type": ["null", { "type": "map", "values": ["null", "string"] }], "default": null },
        { "name": "field_array",        "type": ["null", { "type": "array", "items": "string" }], "default": null },
        { "name": "field_array_of_map", "type": ["null", { "type": "array", "items": { "type": "map", "values": ["null", "string"] }}], "default": null }
    ]
}

Таблица events Hive (формат OR C) создается как: 

CREATE TABLE `events`(
  `id`                  string,
  ......
  `field_map`           map<string,string>,
  `field_array`         array<string>,
  `field_array_of_map`  array<map<string,string>>
)
CLUSTERED BY(id) INTO 9 BUCKETS
STORED AS ORC
TBLPROPERTIES ('transactional'='true');

Поля с array<string>, map<string, string>, array<map<string, string>> типов, они неправильно сохранены в таблице Hive.

Когда в Билайне выдается запрос SELECT, они показывают: 

field_map               {"org.apache.spark.sql.catalyst.expressions.UnsafeMapData@101c5674":null}
field_array             ["org.apache.spark.sql.catalyst.expressions.UnsafeArrayData@6b5730c2"]
field_array_of_map      [{"org.apache.spark.sql.catalyst.expressions.UnsafeArrayData@ca82f1a4":null}]

С https://github.com/hortonworks-spark/spark-llap, это упоминает, что Array тип поддерживается, а Map - нет. Есть идеи как правильно сохранить Array? Любое решение для Map типа?

...