[HIVE-20377] Hive Kafka Storage Handler

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 4.0.0
• Fix Version/s: 4.0.0-alpha-1
• Component/s: kafka integration
• Assignee: Slim Bouguerra
• Reporter: Slim Bouguerra
• Created: 13/Aug/18 16:37
• Updated: 17/Nov/22 08:55
• Resolved: 05/Sep/18 05:22

Description

Goal

• Read streaming data form Kafka queue as an external table.
• Allow streaming navigation by pushing down filters on Kafka record partition id, offset and timestamp.
• Insert streaming data form Kafka to an actual Hive internal table, using CTAS statement.

  Example

  Create the external table

   
  CREATE EXTERNAL TABLE kafka_table (`timestamp` timestamp, page string, `user` string, language string, added int, deleted int, flags string,comment string, namespace string)
  STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
  TBLPROPERTIES 
  ("kafka.topic" = "wikipedia", 
  "kafka.bootstrap.servers"="brokeraddress:9092",
  "kafka.serde.class"="org.apache.hadoop.hive.serde2.JsonSerDe");
  

  Kafka Metadata

  In order to keep track of Kafka records the storage handler will add automatically the Kafka row metadata eg partition id, record offset and record timestamp.

  DESCRIBE EXTENDED kafka_table
  
  timestamp              	timestamp           	from deserializer   
  page                	string              	from deserializer   
  user                	string              	from deserializer   
  language            	string              	from deserializer   
  country             	string              	from deserializer   
  continent           	string              	from deserializer   
  namespace           	string              	from deserializer   
  newpage             	boolean             	from deserializer   
  unpatrolled         	boolean             	from deserializer   
  anonymous           	boolean             	from deserializer   
  robot               	boolean             	from deserializer   
  added               	int                 	from deserializer   
  deleted             	int                 	from deserializer   
  delta               	bigint              	from deserializer   
  __partition         	int                 	from deserializer   
  __offset            	bigint              	from deserializer   
  __timestamp         	bigint              	from deserializer   
  
  

Filter push down.

Newer Kafka consumers 0.11.0 and higher allow seeking on the stream based on a given offset. The proposed storage handler will be able to leverage such API by pushing down filters over metadata columns, namely __partition (int), __offset(long) and __timestamp(long)
For instance Query like

 
select `__offset` from kafka_table where (`__offset` < 10 and `__offset`>3 and `__partition` = 0) or (`__partition` = 0 and `__offset` < 105 and `__offset` > 99) or (`__offset` = 109);

Will result on a scan of partition 0 only then read only records between offset 4 and 109.

With timestamp seeks

The seeking based on the internal timestamps allows the handler to run on recently arrived data, by doing

select count(*) from kafka_table where `__timestamp` >  1000 * to_unix_timestamp(CURRENT_TIMESTAMP - interval '20' hours) ;

This allows for implicit relationships between event timestamps and kafka timestamps to be expressed in queries (i.e event_timestamp is always < than kafka __timestamp and kafka __timestamp is never > 15 minutes from event etc).

More examples with Avro

CREATE EXTERNAL TABLE wiki_kafka_avro_table
STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
TBLPROPERTIES
("kafka.topic" = "wiki_kafka_avro_table",
"kafka.bootstrap.servers"="localhost:9092",
"kafka.serde.class"="org.apache.hadoop.hive.serde2.avro.AvroSerDe",
'avro.schema.literal'='{
  "type" : "record",
  "name" : "Wikipedia",
  "namespace" : "org.apache.hive.kafka",
  "version": "1",
  "fields" : [ {
    "name" : "isrobot",
    "type" : "boolean"
  }, {
    "name" : "channel",
    "type" : "string"
  }, {
    "name" : "timestamp",
    "type" : "string"
  }, {
    "name" : "flags",
    "type" : "string"
  }, {
    "name" : "isunpatrolled",
    "type" : "boolean"
  }, {
    "name" : "page",
    "type" : "string"
  }, {
    "name" : "diffurl",
    "type" : "string"
  }, {
    "name" : "added",
    "type" : "long"
  }, {
    "name" : "comment",
    "type" : "string"
  }, {
    "name" : "commentlength",
    "type" : "long"
  }, {
    "name" : "isnew",
    "type" : "boolean"
  }, {
    "name" : "isminor",
    "type" : "boolean"
  }, {
    "name" : "delta",
    "type" : "long"
  }, {
    "name" : "isanonymous",
    "type" : "boolean"
  }, {
    "name" : "user",
    "type" : "string"
  }, {
    "name" : "deltabucket",
    "type" : "double"
  }, {
    "name" : "deleted",
    "type" : "long"
  }, {
    "name" : "namespace",
    "type" : "string"
  } ]
}'
);

Issue Links

• is related to: HIVE-20561 Use the position of the Kafka Consumer to track progress instead of Consumer Records offsets
• is required by: HIVE-20486 Kafka: Use Row SerDe + vectorization
• links to: Design Doc
• links to: RB Link