湖倉一體電商項目（十一）：

#頭條創作挑戰賽#

編寫寫入DWS層業務代碼

DWS層主要是存放大寬表資料，此業務中主要是針對Kafka topic “KAFKA-DWD-BROWSE-LOG-TOPIC”中使用者浏覽商品日志資料關聯HBase中“ODS_PRODUCT_CATEGORY”商品分類表與“ODS_PRODUCT_INFO”商品表次元資料擷取浏覽商品主題大寬表。

Flink在讀取Kafka 使用者浏覽商品資料與HBase中次元資料進行關聯時采用了Redis做緩存，這樣可以加快處理資料的速度。擷取使用者主題寬表之後，将資料寫入到Iceberg-DWS層中，另外将寬表資料結果寫入到Kafka 中友善後期做實時統計分析。

一、代碼編寫

具體代碼參照“ProduceBrowseLogToDWS.scala”，大體代碼邏輯如下：

object ProduceBrowseLogToDWS {

  private val hbaseDimProductCategoryTbl: String = ConfigUtil.HBASE_DIM_PRODUCT_CATEGORY
  private val hbaseDimProductInfoTbl: String = ConfigUtil.HBASE_DIM_PRODUCT_INFO
  private val kafkaDwsBrowseLogWideTopic: String = ConfigUtil.KAFKA_DWS_BROWSE_LOG_WIDE_TOPIC
  private val kafkaBrokers: String = ConfigUtil.KAFKA_BROKERS

  def main(args: Array[String]): Unit = {
    //1.準備環境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val tblEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    env.enableCheckpointing(5000)
    import org.apache.flink.streaming.api.scala._

    /**
      * 1.需要預先建立 Catalog
      * 建立Catalog,建立表需要在Hive中提前建立好，不在代碼中建立，因為在Flink中建立iceberg表不支援create table if not exists ...文法
      */
    tblEnv.executeSql(
      """
        |create catalog hadoop_iceberg with (
        | 'type'='iceberg',
        | 'catalog-type'='hadoop',
        | 'warehouse'='hdfs://mycluster/lakehousedata'
        |)
      """.stripMargin)

    /**
      * 2.建立 Kafka Connector,連接配接消費Kafka dwd中資料
      *  {
      *   "browseProductCode": "BviQsxHtxC",
      *   "browseProductTpCode": "282",
      *   "userIp": "5.189.85.33",
      *   "obtainPoints": "38",
      *   "userId": "uid250775",
      *   "frontProductUrl": "https:///swdOX/ruh",
      *   "kafka_dwd_topic": "KAFKA-DWD-BROWSE-LOG-TOPIC",
      *   "logTime": "1647067452241",
      *   "browseProductUrl": "https:///57/zB4oF"
      *  }
      */
    tblEnv.executeSql(
      """
        |create table kafka_dwd_browse_log_tbl (
        |   logTime string,
        |   userId string,
        |   userIp string,
        |   frontProductUrl string,
        |   browseProductUrl string,
        |   browseProductTpCode string,
        |   browseProductCode string,
        |   obtainPoints string
        |) with (
        | 'connector' = 'kafka',
        | 'topic' = 'KAFKA-DWD-BROWSE-LOG-TOPIC',
        | 'properties.bootstrap.servers'='node1:9092,node2:9092,node3:9092',
        | 'scan.startup.mode'='earliest-offset', --也可以指定 earliest-offset 、latest-offset
        | 'properties.group.id' = 'my-group-id',
        | 'format' = 'json'
        |)
      """.stripMargin)

    val browseLogTbl:Table = tblEnv.sqlQuery(
      """
        | select logTime,userId,userIp,frontProductUrl,browseProductUrl,browseProductTpCode,browseProductCode,obtainPoints from kafka_dwd_browse_log_tbl
      """.stripMargin)


    //3.将Row 類型資料轉換成對象類型操作,友善與次元資料進行關聯
    val browseLogDS: DataStream[BrowseLog] = tblEnv.toAppendStream[Row](browseLogTbl).map(row=>{
      val logTime: String = row.getField(0).toString//浏覽日志時間
      val userId: String = row.getField(1).toString//使用者編号
      val userIp: String = row.getField(2).toString//浏覽IP位址
      val frontProductUrl: String = row.getField(3).toString//跳轉前URL位址，有為null，有的不為null
      val browseProductUrl: String = row.getField(4).toString//浏覽商品URL
      val browseProductTpCode: String = row.getField(5).toString//浏覽商品二級分類
      val browseProductCode: String = row.getField(6).toString//浏覽商品編号
      val obtainPointsstring: String = row.getField(7).toString//浏覽商品所獲積分
      BrowseLog(logTime,userId,userIp,frontProductUrl,browseProductUrl,browseProductTpCode,browseProductCode,obtainPointsstring)
    })

    //4.設定Sink 到Kafka 資料輸出到側輸出流标記
    val kafkaDataTag = new OutputTag[JSONObject]("kafka_data")

    //5.連接配接phoenix 庫查詢HBase資料組織Browse寬表
    val browseLogWideInfoDS: DataStream[BrowseLogWideInfo] = browseLogDS.process(new ProcessFunction[BrowseLog,BrowseLogWideInfo] {

      var conn: Connection = _
      var pst: PreparedStatement = _
      var rs: ResultSet = _

      //建立Phoenix 連接配接
      override def open(parameters: Configuration): Unit = {
        //連接配接Phoenix
        println(s"連接配接Phoenix ... ...")
        conn = DriverManager.getConnection(ConfigUtil.PHOENIX_URL)
      }

      override def processElement(browseLog: BrowseLog, context: ProcessFunction[BrowseLog, BrowseLogWideInfo]#Context, collector: Collector[BrowseLogWideInfo]): Unit ={
        //最終傳回的json 對象
        val jsonObj = new JSONObject()
        jsonObj.put("log_time", browseLog.logTime)
        jsonObj.put("user_id", browseLog.userId)
        jsonObj.put("user_ip", browseLog.userIp)
        jsonObj.put("front_product_url", browseLog.frontProductUrl)
        jsonObj.put("browse_product_url", browseLog.browseProductUrl)
        jsonObj.put("browse_product_tpcode", browseLog.browseProductTpCode) //商品類型id
        jsonObj.put("browse_product_code", browseLog.browseProductCode)//商品id
        jsonObj.put("obtain_points", browseLog.obtainPoints)


        //根據浏覽商品類型id : browse_product_tpcode 從Redis緩存中讀取 DIM_PRODUCT_CATEGORY - 商品類别表
        val productCategoryRedisCacheInfo: String = MyRedisUtil.getInfoFromRedisCache(hbaseDimProductCategoryTbl, browseLog.browseProductTpCode)

        //根據浏覽商品id : browse_product_code 從Redis緩存中讀取 DIM_PRODUCT_INFO - 商品基本資訊表
        val productInfoRedisCacheInfo: String = MyRedisUtil.getInfoFromRedisCache(hbaseDimProductInfoTbl, browseLog.browseProductCode)

        //商品種類資料如果 Redis 緩存中沒有則讀取phoenix擷取，有則直接從緩存中擷取
        if (MyStringUtil.isEmpty(productCategoryRedisCacheInfo)) {
          //說明緩存中沒有資料，從phoenix中查詢
          println("連接配接Phoenix查詢  DIM_PRODUCT_CATEGORY - 商品類别表 次元資料")
          val sql =
            s"""
               |SELECT
               | b.id as first_category_id,
               | b.name AS first_category_name,
               | a.id as second_category_id,
               | a.name AS second_category_name
               |FROM DIM_PRODUCT_CATEGORY a JOIN DIM_PRODUCT_CATEGORY b ON a.p_id = b.id where a.id = '${browseLog.browseProductTpCode}'
            """.stripMargin

          println("phoenix 執行SQL 如下： "+sql)
          pst = conn.prepareStatement(sql)
          rs = pst.executeQuery()

          //準備 向Redis 中寫入 DIM_PRODUCT_CATEGORY - 商品類别表 的json對象
          val dimProductCategroyRedisJsonObj = new JSONObject()
          while (rs.next()) {
            dimProductCategroyRedisJsonObj.put("first_category_id", rs.getString("first_category_id"))
            dimProductCategroyRedisJsonObj.put("first_category_name", rs.getString("first_category_name"))
            dimProductCategroyRedisJsonObj.put("second_category_id", rs.getString("second_category_id"))
            dimProductCategroyRedisJsonObj.put("second_category_name", rs.getString("second_category_name"))

            //将商品種類資訊存入Redis緩存，向Redis中設定資料緩存
            MyRedisUtil.setRedisDimCache(hbaseDimProductCategoryTbl, browseLog.browseProductTpCode, dimProductCategroyRedisJsonObj.toString)

            //将json 加入到總傳回結果的Json中
            CommonUtil.AddAttributeToJson(jsonObj, dimProductCategroyRedisJsonObj)
          }

        }else{
          //Redis中查詢到了資料，從redis 中擷取 json 資訊設定在最終結果中
          println("DIM_PRODUCT_CATEGORY - 商品類别表  從Redis中擷取到緩存處理")
          CommonUtil.AddAttributeToJson(jsonObj, JSON.parseObject(productCategoryRedisCacheInfo))

        }

        //商品資訊資料如果 Redis 緩存中沒有則讀取phoenix擷取，有則直接從緩存中擷取
        if (MyStringUtil.isEmpty(productInfoRedisCacheInfo)) {
          //說明緩存中沒有資料，從phoenix中查詢
          println("連接配接Phoenix查詢  DIM_PRODUCT_INFO - 商品基本資訊表 次元資料")
          val sql =
            s"""
               |SELECT
               | product_id,
               | product_name
               |FROM DIM_PRODUCT_INFO where product_id = '${browseLog.browseProductCode}'
            """.stripMargin

          println("phoenix 執行SQL 如下： "+sql)
          pst = conn.prepareStatement(sql)
          rs = pst.executeQuery()

          //準備 向Redis 中寫入 DIM_PRODUCT_INFO - 商品基本資訊表 的json對象
          val dimProductInfoRedisJsonObj = new JSONObject()
          while (rs.next()) {
            dimProductInfoRedisJsonObj.put("product_id", rs.getString("product_id"))
            dimProductInfoRedisJsonObj.put("product_name", rs.getString("product_name"))

            //将商品種類資訊存入Redis緩存，向Redis中設定資料緩存
            MyRedisUtil.setRedisDimCache(hbaseDimProductInfoTbl, browseLog.browseProductCode, dimProductInfoRedisJsonObj.toString)

            //将json 加入到總傳回結果的Json中
            CommonUtil.AddAttributeToJson(jsonObj, dimProductInfoRedisJsonObj)
          }

        }else{
          //Redis中查詢到了資料，從redis 中擷取 json 資訊設定在最終結果中
          println("DIM_PRODUCT_INFO - 商品基本資訊表  從Redis中擷取到緩存處理")
          CommonUtil.AddAttributeToJson(jsonObj, JSON.parseObject(productInfoRedisCacheInfo))
        }

        //準備向Kafka 中存儲的資料json 對象
        context.output(kafkaDataTag,jsonObj)

        //最終傳回 jsonObj,此時jsonObj包含了所有json 資訊
        /**
          *  {
          *   "first_category_id": "30",
          *   "user_ip": "195.134.35.113",
          *   "obtain_points": "0",
          *   "product_name": "扭扭車",
          *   "log_time": "2022-03-17 16:22:09",
          *   "browse_product_tpcode": "30000",
          *   "front_product_url": "https://0BZ/7N/qVIap",
          *   "first_category_name": "玩具樂器",
          *   "user_id": "uid786601",
          *   "browse_product_code": "xA4cfipkdl",
          *   "product_id": "xA4cfipkdl",
          *   "second_category_id": "30000",
          *   "browse_product_url": "https://DU6S2wiT/n/l3E",
          *   "second_category_name": "童車童床"
          *  }
          */
        collector.collect(BrowseLogWideInfo(jsonObj.getString("log_time").split(" ")(0),jsonObj.getString("user_id"),jsonObj.getString("user_ip"),
          jsonObj.getString("product_name"),jsonObj.getString("front_product_url"),jsonObj.getString("browse_product_url"),jsonObj.getString("first_category_name"),
          jsonObj.getString("second_category_name"),jsonObj.getString("obtain_points")))

      }

      override def close(): Unit = {
        rs.close()
        pst.close()
        conn.close()
      }
    })

    /**
      * 6.将清洗完的資料存入Iceberg 表中
      * 将寬表轉換成表存儲在 iceberg - DWS 層 DWS_BROWSE_INFO ,
      */
    val table: Table = tblEnv.fromDataStream(browseLogWideInfoDS)
    tblEnv.executeSql(
      s"""
         |insert into hadoop_iceberg.icebergdb.DWS_BROWSE_INFO
         |select
         | log_time,
         | user_id,
         | user_ip,
         | product_name,
         | front_product_url,
         | browse_product_url,
         | first_category_name,
         | second_category_name,
         | obtain_points
         | from ${table}
      """.stripMargin)


    //7.同時将結果存儲在Kafka  KAFKA-DWS-BROWSE-LOG-WIDE-TOPIC topic中
    /**
      * 将以上資料寫入到Kafka 各自DWD 層topic中，這裡不再使用SQL方式，而是直接使用DataStream代碼方式 Sink 到各自的DWD層代碼中
      */
    val props = new Properties()
    props.setProperty("bootstrap.servers",kafkaBrokers)

    browseLogWideInfoDS.getSideOutput(kafkaDataTag).addSink(new FlinkKafkaProducer[JSONObject](kafkaDwsBrowseLogWideTopic,new KafkaSerializationSchema[JSONObject] {
      override def serialize(jsonObj: JSONObject, timestamp: java.lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
        new ProducerRecord[Array[Byte], Array[Byte]](kafkaDwsBrowseLogWideTopic,null,jsonObj.toString.getBytes())
      }
    },props,FlinkKafkaProducer.Semantic.AT_LEAST_ONCE))

    env.execute()
  }
}           

二、建立Iceberg-DWS層表

代碼在執行之前需要在Hive中預先建立對應的Iceberg表，建立Icebreg表方式如下：

1、在Hive中添加Iceberg表格式需要的包

啟動HDFS叢集，node1啟動Hive metastore服務，在Hive用戶端啟動Hive添加Iceberg依賴包：

#node1節點啟動Hive metastore服務
[root@node1 ~]# hive --service metastore &

#在hive用戶端node3節點加載兩個jar包
add jar /software/hive-3.1.2/lib/iceberg-hive-runtime-0.12.1.jar;
add jar /software/hive-3.1.2/lib/libfb303-0.9.3.jar;           

2、建立Iceberg表

這裡建立Iceberg-DWS表有“DWS_BROWSE_INFO”,建立語句如下：

CREATE TABLE DWS_BROWSE_INFO (
log_time string,
user_id string,
user_ip string,
product_name string,
front_product_url string,
browse_product_url string,
first_category_name string,
second_category_name string,
obtain_points string
)STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' 
LOCATION 'hdfs://mycluster/lakehousedata/icebergdb/DWS_BROWSE_INFO/' 
TBLPROPERTIES ('iceberg.catalog'='location_based_table',
'write.metadata.delete-after-commit.enabled'= 'true',
'write.metadata.previous-versions-max' = '3'
);           

三、代碼測試

以上代碼編寫完成後，代碼執行測試步驟如下：

1、在Kafka中建立對應的topic

#在Kafka 中建立 KAFKA-DWS-BROWSE-LOG-WIDE-TOPIC topic
./kafka-topics.sh --zookeeper node3:2181,node4:2181,node5:2181 --create --topic KAFKA-DWS-BROWSE-LOG-WIDE-TOPIC --partitions 3 --replication-factor 3

#監控以上topic資料
[root@node1 bin]# ./kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic KAFKA-DWS-BROWSE-LOG-WIDE-TOPIC           

2、将代碼中消費Kafka資料改成從頭開始消費

代碼中Kafka Connector中屬性“scan.startup.mode”設定為“earliest-offset”，從頭開始消費資料。

這裡也可以不設定從頭開始消費Kafka資料，而是直接啟動向日志采集接口模拟生産日志代碼“RTMockUserLogData.java”，需要啟動日志采集接口及Flume。

3、執行代碼，檢視對應結果

以上代碼執行後在，在對應的Kafka “KAFKA-DWS-BROWSE-LOG-WIDE-TOPIC” topic中都有對應的資料。在Iceberg-DWS層中對應的表中也有資料。

Kafka中結果如下：

Iceberg-DWS層表”DWS_BROWSE_INFO”中的資料如下：

四、架構圖