基于Spark Structured Streaming实现Kafka到HBase传输，需设置自定义时间戳

基于你给出的技术栈（Spark 2.2 + HDP 2.6.3生态 + 自定义编译的shc-core-1.1.2-2.2-s_2.11-SNAPSHOT.jar），我来分享一套可行的实时ETL实现方案，重点覆盖Kafka到HBase的Structured Streaming传输以及自定义时间戳的设置：

我们会通过Spark Structured Streaming读取Kafka中推送的客户对象变更数据，解析后使用对象自带的LastUpdateDate字段作为自定义事件时间戳，最后通过你编译的HBase Sink Provider将数据写入HBase表中，全程保证流式处理的可靠性与实时性。

1. 定义客户对象Schema并读取Kafka数据源 #

首先需要定义客户对象的结构化Schema，用于解析Kafka中传递的JSON格式数据：

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

// 匹配客户对象的字段定义
val customerSchema = StructType(Seq(
  StructField("Name", StringType, nullable = false),
  StructField("Surname", StringType, nullable = false),
  StructField("Status", StringType),
  StructField("Email", StringType),
  StructField("Version", IntegerType),
  StructField("LastUpdateDate", StringType, nullable = false) // 假设时间格式为"yyyy-MM-dd HH:mm:ss"
))

// 读取Kafka流式数据
val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-broker1:9092,your-kafka-broker2:9092")
  .option("subscribe", "customer-change-topic")
  .option("startingOffsets", "earliest") // 首次运行可从最早偏移量开始消费
  .load()

2. 解析数据并设置自定义时间戳 #

接下来解析Kafka的value字段为结构化DataFrame，并将LastUpdateDate转换为Spark Timestamp类型作为自定义事件时间戳（这一步是核心需求的关键）：

// 解析JSON数据并提取字段
val customerDF = kafkaStream
  .selectExpr("CAST(value AS STRING)") // 将Kafka的二进制value转为字符串
  .select(from_json($"value", customerSchema).as("customer"))
  .select("customer.*")

// 将业务字段LastUpdateDate转为事件时间戳，格式需与实际数据匹配
val customerWithEventTimeDF = customerDF
  .withColumn("eventTime", to_timestamp($"LastUpdateDate", "yyyy-MM-dd HH:mm:ss"))

如果后续需要基于事件时间做窗口聚合或水位线（Watermark）控制，可以额外添加：

// 可选：设置水位线处理延迟数据
val customerWithWatermarkDF = customerWithEventTimeDF
  .withWatermark("eventTime", "10 minutes")

3. 配置HBase映射并写入HBase #

使用你编译的shc-core包，通过指定catalog映射客户字段到HBase的列族与列，同时设置必要的流式作业参数：

// 定义HBase表的catalog映射，需提前创建好HBase表与列族
val hbaseCatalog =
  s"""{
     |  "table":{"namespace":"default", "name":"customer"},
     |  "rowkey":"rowkey",
     |  "columns":{
     |    "Name":{"cf":"info", "col":"name", "type":"string"},
     |    "Surname":{"cf":"info", "col":"surname", "type":"string"},
     |    "Status":{"cf":"info", "col":"status", "type":"string"},
     |    "Email":{"cf":"info", "col":"email", "type":"string"},
     |    "Version":{"cf":"info", "col":"version", "type":"int"},
     |    "LastUpdateDate":{"cf":"info", "col":"last_update_date", "type":"string"},
     |    "rowkey":{"cf":"rowkey", "col":"rowkey", "type":"string"}
     |  }
     |}""".stripMargin

// 生成HBase RowKey（这里用Name+Surname作为唯一标识，可根据实际业务调整为客户ID等）
val customerWithRowKeyDF = customerWithEventTimeDF
  .withColumn("rowkey", concat($"Name", lit("_"), $"Surname"))

// 启动流式写入HBase的作业
val streamingQuery = customerWithRowKeyDF.writeStream
  .format("org.apache.spark.sql.execution.datasources.hbase")
  .option("catalog", hbaseCatalog)
  .option("hbase.configuration", "/etc/hbase/conf/hbase-site.xml") // 指向HDP集群的HBase配置文件
  .option("checkpointLocation", "/user/spark/checkpoint/customer-etl") // 必须指定，用于故障恢复
  .outputMode("append") // 因为每个变更都是完整对象，append模式即可
  .start()

// 保持作业运行
streamingQuery.awaitTermination()

• 依赖包管理：确保自定义编译的shc-core-1.1.2-2.2-s_2.11-SNAPSHOT.jar已添加到Spark的classpath中，或在提交作业时通过--jars参数指定。
• HBase表准备：提前在HBase中创建customer表，确保列族info存在（对应catalog中的配置）。
• 时间格式匹配：to_timestamp的格式参数必须与LastUpdateDate的实际格式完全一致，否则会生成null值。
• 流式作业可靠性：checkpointLocation必须指向HDFS路径，保证作业故障重启后能从断点继续消费。
• 性能调优：根据Kafka消息量与HBase写入能力，调整spark.streaming.kafka.maxRatePerPartition控制消费速度，同时可以调整Spark的spark.sql.shuffle.partitions参数优化并行度。

内容的提问来源于stack exchange，提问作者ser0t0nin