When creating a Hive table with Druid storage handler CREATE TABLE MyDruidTable ( `__time` timestamp , `dim1` string , `meas` int) STORED BY 'org.apache.hadoop.hive.druid.DruidStorageHandler' and executing this Spark code import org.apache.spark.sql.SaveMode val my_df = spark.read.parquet("<source>").withColumn("__time", (col("crdt")/1000).cast("timestamp")).limit(2) my_df .write.mode(SaveMode.Overwrite).saveAsTable("MyDruidTable") I get following error stack trace. java.util.NoSuchElementException: None.get at scala.None$.get(Option.scala:347) at scala.None$.get(Option.scala:345) at org.apache.spark.sql.execution.datasources.PreprocessTableCreation$$anonfun$apply$2.applyOrElse(rules.scala:110) at org.apache.spark.sql.execution.datasources.PreprocessTableCreation$$anonfun$apply$2.applyOrElse(rules.scala:75) at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267) at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267) at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70) at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266) at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256) at org.apache.spark.sql.execution.datasources.PreprocessTableCreation.apply(rules.scala:75) at org.apache.spark.sql.execution.datasources.PreprocessTableCreation.apply(rules.scala:71) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:85) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:82) at scala.collection.IndexedSeqOptimized$class.foldl(IndexedSeqOptimized.scala:57) at scala.collection.IndexedSeqOptimized$class.foldLeft(IndexedSeqOptimized.scala:66) at scala.collection.mutable.ArrayBuffer.foldLeft(ArrayBuffer.scala:48) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:82) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:74) at scala.collection.immutable.List.foreach(List.scala:381) at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:74) at org.apache.spark.sql.execution.QueryExecution.analyzed$lzycompute(QueryExecution.scala:69) at org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:67) at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:50) at org.apache.spark.sql.execution.QueryExecution.withCachedData$lzycompute(QueryExecution.scala:73) at org.apache.spark.sql.execution.QueryExecution.withCachedData(QueryExecution.scala:72) at org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:78) at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:78) at org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:84) at org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:80) at org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:89) at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:89) at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:92) at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:92) at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:609) at org.apache.spark.sql.DataFrameWriter.createTable(DataFrameWriter.scala:419) at org.apache.spark.sql.DataFrameWriter.saveAsTable(DataFrameWriter.scala:398) at org.apache.spark.sql.DataFrameWriter.saveAsTable(DataFrameWriter.scala:354) ... 50 elided Doesn't Spark support Druid? Or is there a way to troubleshoot? We're using Spark with LLAP based on https://community.hortonworks.com/articles/101181/rowcolumn-level-security-in-sql-for-apache-spark-2.html ... View more

1. Create Kafka Connect docker run -d \ --name=connect-distributed \ --net=host \ -e CONNECT_BOOTSTRAP_SERVERS=<kafka cluster> \ -e CONNECT_REST_PORT=8082 \ -e CONNECT_GROUP_ID="ingest-hdfs-from-kafka" \ -e CONNECT_CONFIG_STORAGE_TOPIC="kafka-hdfs-config" \ -e CONNECT_OFFSET_STORAGE_TOPIC="kafka-hdfs-offsets" \ -e CONNECT_STATUS_STORAGE_TOPIC="kafka-hdfs-status" \ -e CONNECT_KEY_CONVERTER="org.apache.kafka.connect.storage.StringConverter" \ -e CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE="false" \ -e CONNECT_VALUE_CONVERTER="org.apache.kafka.connect.storage.StringConverter" \ -e CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE="false" \ -e CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \ -e CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \ -e CONNECT_REST_ADVERTISED_HOST_NAME=<host> \ -v /gkn/sw/kafka-connect/etc/hadoop:/etc/hadoop:ro \ -v /gkn/sw/kafka-connect/etc/keytab:/etc/keytab:ro \ -v /gkn/sw/kafka-connect/etc/krb5.conf:/etc/krb5.conf:ro \ confluentinc/cp-kafka-connect:3.3.0 2. Create HDFS Sink, default in avro format curl -X POST -H "Content-Type: application/json" --data '{"name": "hdfs-sink", "config": {"connector.class":"io.confluent.connect.hdfs.HdfsSinkConnector", "tasks.max":"3", "topics":"SENSORS", "hadoop.conf.dir":"/etc/hadoop", "hdfs.url":"hdfs://<hadoop>", "flush.size":"10000000", "partition.duration.ms": "300000", "partitioner.class":"io.confluent.connect.hdfs.partitioner.DailyPartitioner", "rotate.schedule.interval.ms": "300000", "topics.dir":"/myfolder/ingest/kafka/", "logs.dir":"/myfolder/ingest/logs/", "hdfs.authentication.kerberos":"true", "connect.hdfs.principal":"<principal>", "connect.hdfs.keytab":"<keytab>", "hdfs.namenode.principal":"", "locale":"en-us", "timezone":"GMT" }}' http://<kafka-connect>:28082/connectors 3. Extract schema from Avro java -jar avro-tools-1.8.2.jar getschema sensors+0+0060650523+0060651522.avro > environment.avsc The result is this: [ "null", "bytes" ] 4. Create Hive table CREATE EXTERNAL TABLE SENSORDATA ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe' STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat' LOCATION '/folder/SENSORDATA' TBLPROPERTIES ('avro.schema.literal'= '{"name": "parent", "type": "record", "fields": [ {"name": "children", "type": {"type": "array", "items": {"name": "child", "type": "record", "fields": [ {"name": "id", "type": "string"}, {"name": "topics", "type": "string"}, {"name": "position", "type": "string"}, {"name": "category", "type": "string"}, {"name": "value", "type": "float"}, {"name": "unit", "type": "string"}, {"name": "timestamp", "type": "string"} ] }}} ]}' ); or CREATE EXTERNAL TABLE SENSORDATA (a string, i struct<id:STRING, topics:STRING, position:STRING, category:STRING, value:FLOAT, unit:STRING, ts:STRING>) STORED AS avro LOCATION '/folder/SENSORDATA' Then select * from SENSORDATA limit 10; Leads to this error org.apache.hive.service.cli.HiveSQLException: java.io.IOException: org.apache.avro.AvroTypeException: Found string, expecting parent Do you have any suggestion? Or is there a better way to stream JSON data. Note: Schema definition Kafka producer is not feasible. Any guidance is appreciated. ... View more