我正在使用Kafka Consumer API將所有數據從Kafka主題復制到Hive表 . 為此，我使用HDFS作為中間步驟 . 我使用唯一的組ID并將偏移重置為“最早”，以便從頭開始獲取所有數據，并在執行后忽略提交 . 然后我遍歷Kafka主題中的記錄，并將每條記錄保存到HDFS中的臨時文件中 . 然后我使用Spark從HDFS讀取數據，然后使用日期作為文件名將其保存到Parquet文件中 . 然后，我在Hive表中創建一個帶日期的分區，最后在Parquet中將文件作為分區加載到Hive中 .
正如您在下面的代碼中看到的，我使用了幾個中間步驟，這使得我的代碼遠非最佳 . 這是從Kafka主題復制所有數據的最佳推薦方法嗎？我做了一些研究，到目前為止，這是我設法開始工作的變通方法，但是，隨著記錄數量每天增加，我的執行時間達到了可容忍的極限(從2分鐘變為6分鐘到6分鐘)周) .
代碼在這里：
def start( lowerDate: String, upperDate: String )={
// Configurations for kafka consumer
val conf = ConfigFactory.parseResources("properties.conf")
val brokersip = conf.getString("enrichment.brokers.value")
val topics_in = conf.getString("enrichment.topics_in.value")
// Crea la sesion de Spark
val spark = SparkSession
.builder()
.master("yarn")
.appName("ParaTiUserXY")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
import spark.implicits._
val properties = new Properties
properties.put("key.deserializer", classOf[StringDeserializer])
properties.put("value.deserializer", classOf[StringDeserializer])
properties.put("bootstrap.servers", brokersip)
properties.put("auto.offset.reset", "earliest")
properties.put("group.id", "ParaTiUserXYZZ12345")
//Schema para transformar los valores del topico de Kafka a JSON
val my_schema = new StructType()
.add("longitudCliente", StringType)
.add("latitudCliente", StringType)
.add("dni", StringType)
.add("alias", StringType)
.add("segmentoCliente", StringType)
.add("timestampCliente", StringType)
.add("dateCliente", StringType)
.add("timeCliente", StringType)
.add("tokenCliente", StringType)
.add("telefonoCliente", StringType)
val consumer = new KafkaConsumer[String, String](properties)
consumer.subscribe( util.Collections.singletonList("parati_rt_geoevents") )
val fs = {
val conf = new Configuration()
FileSystem.get(conf)
}
val temp_path:Path = new Path("hdfs:///tmp/s70956/tmpstgtopics")
if( fs.exists(temp_path)){
fs.delete(temp_path, true)
}
while(true)
{
val records=consumer.poll(100)
for (record
val data = record.value.toString
//println(data)
val dataos: FSDataOutputStream = fs.create(temp_path)
val bw: BufferedWriter = new BufferedWriter( new OutputStreamWriter(dataos, "UTF-8"))
bw.append(data)
bw.close
val data_schema = spark.read.schema(my_schema).json("hdfs:///tmp/s70956/tmpstgtopics")
val fechaCliente = data_schema.select("dateCliente").first.getString(0)
if( fechaCliente < upperDate && fechaCliente >= lowerDate){
data_schema.select("longitudCliente", "latitudCliente","dni", "alias",
"segmentoCliente", "timestampCliente", "dateCliente", "timeCliente",
"tokenCliente", "telefonoCliente")
.coalesce(1).write.mode(SaveMode.Append).parquet("/desa/landing/parati/xyuser/" + fechaCliente)
}
else if( fechaCliente < lowerDate){
//
}
else if( fechaCliente >= upperDate){
break;
}
}
}
consumer.close()
}
