数据湖技术之iceberg(十)Structured Streaming实时写入Iceberg

南墨2年前技术文章718

目前Spark中Structured Streaming只支持实时向Iceberg中写入数据,不支持实时从Iceberg中读取数据,下面案例我们将使用Structured Streaming从Kafka中实时读取数据,然后将结果实时写入到Iceberg中。

一、创建Kafka topic

启动Kafka集群,创建“kafka-iceberg-topic”

[root@node1 bin]# ./kafka-topics.sh  --zookeeper node3:2181,node4:2181,node5:2181  --create  --topic kafka-iceberg-topic  --partitions 3 --replication-factor 3

二、编写向Kafka生产数据代码

/**

  * 向Kafka中写入数据

  */

object WriteDataToKafka {

  def main(args: Array[String]): Unit = {

    val props = new Properties()

    props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092")

    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

 

    val producer = new KafkaProducer[String,String](props)

    var counter = 0

    var keyFlag = 0

    while(true){

      counter +=1

      keyFlag +=1

      val content: String = userlogs()

      producer.send(new ProducerRecord[String, String]("kafka-iceberg-topic", content))

      //producer.send(new ProducerRecord[String, String]("kafka-iceberg-topic", s"key-$keyFlag", content))

      if(0 == counter%100){

        counter = 0

        Thread.sleep(5000)

      }

    }

    producer.close()

  }

 

  def userlogs()={

    val userLogBuffer = new StringBuffer("")

    val timestamp = new Date().getTime();

    var userID = 0L

    var pageID = 0L

 

    //随机生成的用户ID

    userID = Random.nextInt(2000)

 

    //随机生成的页面ID

    pageID =  Random.nextInt(2000);

 

    //随机生成Channel

    val channelNames = Array[String]("Spark","Scala","Kafka","Flink","Hadoop","Storm","Hive","Impala","HBase","ML")

    val channel = channelNames(Random.nextInt(10))

 

    val actionNames = Array[String]("View", "Register")

    //随机生成action行为

    val action = actionNames(Random.nextInt(2))

 

    val dateToday = new SimpleDateFormat("yyyy-MM-dd").format(new Date())

    userLogBuffer.append(dateToday)

      .append("\t")

      .append(timestamp)

      .append("\t")

      .append(userID)

      .append("\t")

      .append(pageID)

      .append("\t")

      .append(channel)

      .append("\t")

      .append(action)

    System.out.println(userLogBuffer.toString())

    userLogBuffer.toString()

  }

}

三、编写Structured Streaming读取Kafka数据实时写入Iceberg

object StructuredStreamingSinkIceberg {

  def main(args: Array[String]): Unit = {

    //1.准备对象

    val spark: SparkSession = SparkSession.builder().master("local").appName("StructuredSinkIceberg")

      //指定hadoop catalog,catalog名称为hadoop_prod

      .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")

      .config("spark.sql.catalog.hadoop_prod.type", "hadoop")

      .config("spark.sql.catalog.hadoop_prod.warehouse", "hdfs://mycluster/structuredstreaming")

      .getOrCreate()

//    spark.sparkContext.setLogLevel("Error")

 

    //2.创建Iceberg 表

    spark.sql(

      """

        |create table if not exists hadoop_prod.iceberg_db.iceberg_table (

        | current_day string,

        | user_id string,

        | page_id string,

        | channel string,

        | action string

        |) using iceberg

      """.stripMargin)

 

    val checkpointPath = "hdfs://mycluster/iceberg_table_checkpoint"

    val bootstrapServers = "node1:9092,node2:9092,node3:9092"

    //多个topic 逗号分开

    val topic = "kafka-iceberg-topic"

 

    //3.读取Kafka读取数据

    val df = spark.readStream

      .format("kafka")

      .option("kafka.bootstrap.servers", bootstrapServers)

      .option("auto.offset.reset", "latest")

      .option("group.id", "iceberg-kafka")

      .option("subscribe", topic)

      .load()

 

    import spark.implicits._

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

 

    val resDF = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

      .as[(String, String)].toDF("id", "data")

 

    val transDF: DataFrame = resDF.withColumn("current_day", split(col("data"), "\t")(0))

      .withColumn("ts", split(col("data"), "\t")(1))

      .withColumn("user_id", split(col("data"), "\t")(2))

      .withColumn("page_id", split(col("data"), "\t")(3))

      .withColumn("channel", split(col("data"), "\t")(4))

      .withColumn("action", split(col("data"), "\t")(5))

      .select("current_day", "user_id", "page_id", "channel", "action")

 

    //结果打印到控制台,Default trigger (runs micro-batch as soon as it can)

//    val query: StreamingQuery = transDF.writeStream

//      .outputMode("append")

//      .format("console")

//      .start()

 

    //4.流式写入Iceberg表

    val query = transDF.writeStream

      .format("iceberg")

      .outputMode("append")

      //每分钟触发一次Trigger.ProcessingTime(1, TimeUnit.MINUTES)

      //每10s 触发一次 Trigger.ProcessingTime(1, TimeUnit.MINUTES)

      .trigger(Trigger.ProcessingTime(10, TimeUnit.SECONDS))

      .option("path", "hadoop_prod.iceberg_db.iceberg_table")

      .option("fanout-enabled", "true")

      .option("checkpointLocation", checkpointPath)

      .start()

 

    query.awaitTermination()

 

  }

}

注意:以上代码执行时由于使用的Spark版本为3.1.2,其依赖的Hadoop版本为Hadoop3.2版本,所以需要在本地Window中配置Hadoop3.1.2的环境变量以及将对应的hadoop.dll放入window "C:\Windows\System32"路径下。

Structuerd Streaming向Iceberg实时写入数据有以下几个注意点:

写Iceberg表写出数据支持两种模式:append和complete,append是将每个微批数据行追加到表中。complete是替换每个微批数据内容。

向Iceberg中写出数据时指定的path可以是HDFS路径,可以是Iceberg表名,如果是表名,要预先创建好Iceberg表。

写出参数fanout-enabled指的是如果Iceberg写出的表是分区表,在向表中写数据之前要求Spark每个分区的数据必须排序,但这样会带来数据延迟,为了避免这个延迟,可以设置“fanout-enabled”参数为true,可以针对每个Spark分区打开一个文件,直到当前task批次数据写完,这个文件再关闭。

实时向Iceberg表中写数据时,建议trigger设置至少为1分钟提交一次,因为每次提交都会产生一个新的数据文件和元数据文件,这样可以减少一些小文件。为了进一步减少数据文件,建议定期合并“data files”(参照1.9.6.9)和删除旧的快照(1.9.6.10)。

四、查看Iceberg中数据结果

启动向Kafka生产数据代码,启动向Iceberg中写入数据的Structured Streaming程序,执行以下代码来查看对应的Iceberg结果:

//1.准备对象

val spark: SparkSession = SparkSession.builder().master("local").appName("StructuredSinkIceberg")

  //指定hadoop catalog,catalog名称为hadoop_prod

  .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")

  .config("spark.sql.catalog.hadoop_prod.type", "hadoop")

  .config("spark.sql.catalog.hadoop_prod.warehouse", "hdfs://mycluster/structuredstreaming")

  .getOrCreate()

 

//2.读取Iceberg 表中的数据结果

spark.sql(

  """

    |select * from hadoop_prod.iceberg_db.iceberg_table

  """.stripMargin).show()


相关文章

MySQL运维实战(7.2) MySQL复制server_id相关问题

MySQL运维实战(7.2) MySQL复制server_id相关问题

主库server_id没有设置主库没有设置server_idGot fatal error 1236 from master when&nb...

Oozie安装web页面

Oozie安装web页面

1、查看web页面,页面显示页面显示无法打开oozie的web安装页面,如果需要安装页面需要安装Ext js的lib库2、安装ExtJS 2.2库进入/var/lib/oozie路径中cd  /var...

Clickhouse MergeTree原理(二)—— 表和分区的维护

MergeTree是Clickhouse中最核心的存储引擎。上一篇文章中,我们介绍了MergeTree的基本结构。1、MergeTree由分区(partiton)和part组成。2、Part是Merg...

MySQL运维实战之ProxySQL(9.5)proxysql和MySQL Group Replication配合使用

如果后端MySQL使用了Group Replication,可通过配置mysql_group_replication_hostgroups表来实现高可用mysql_group_replication_...

Go 链表的实现

Go 链表的实现

链表是一种物理存储单元上非连续、非顺序的存储结构,数据元素的逻辑顺序是通过链表中的指针链接次序实现的。链表由一系列结点(链表中每一个元素称为结点)组成,结点可以在运行时动态生成。每个结点包括两个部分:...

PG的多版本并发控制(一)

PG的多版本并发控制(一)

一、 表系统字段几个比较重要概念1.1  tupletuple表示表中的数据行,在MySQL中用row表示。在表数据页中,主要分为普通的数据元祖和TOAST元祖。以下是一个普通数据元祖的结构,主要由三...

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。