<dependencies>
	<dependency>
    	<groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.0.24</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.58</version>
    </dependency>
</dependencies>

Entry  
    Header  
        logfileName [binlog文件名]  
        logfileOffset [binlog position]  
        executeTime [binlog里记录变更发生的时间戳,精确到秒]  
        schemaName   
        tableName  
        eventType [insert/update/delete类型]  
    entryType   [事务头BEGIN/事务尾END/数据ROWDATA]  
    storeValue  [byte数据,可展开，对应的类型为RowChange]  
RowChange
    isDdl       [是否是ddl变更操作，比如create table/drop table]
    sql         [具体的ddl sql]
rowDatas    [具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理]
    beforeColumns [Column类型的数组，变更前的数据字段]
    afterColumns [Column类型的数组，变更后的数据字段]
    Column
    index
    sqlType     [jdbc type]
    name        [column name]
    isKey       [是否为主键]
    updated     [是否发生过变更]
    isNull      [值是否为null]
    value       [具体的内容，注意为string文本]

# canal配置
canal.server.ip=node1
canal.server.port=11111
canal.server.destination=example
canal.server.username=canal
canal.server.password=canal
canal.subscribe.filter=itcast_shop.*

# zookeeper配置
zookeeper.server.ip=hadoop102:2181,hadoop103:2181,hadoop104:2181

# kafka配置
kafka.bootstrap_servers_config=hadoop102:9092,hadoop103:9092,hadoop104:9092
kafka.batch_size_config=1024
kafka.acks=all
kafka.retries=0
kafka.client_id_config=carlos_shop_canal_click
kafka.key_serializer_class_config=org.apache.kafka.common.serialization.StringSerializer
kafka.value_serializer_class_config=com.carlos.canal.protobuf.ProtoBufSerializer
kafka.topic=ods_carlos_shop_mysql

/**
 * 入口
 */
public class Entrance {
    public static void main(String[] args) {
        CanalClient canalClient = new CanalClient();
        canalClient.start();
    }
}

/**
 * Kafka生产者
 */
public class KafkaSender {
    private Properties kafkaProps = new Properties();
    private KafkaProducer<String, RowData> kafkaProducer;

    public KafkaSender() {
        kafkaProps.put("bootstrap.servers", ConfigUtil.kafkaBootstrap_servers_config());
        kafkaProps.put("acks", ConfigUtil.kafkaAcks());
        kafkaProps.put("retries", ConfigUtil.kafkaRetries());
        kafkaProps.put("batch.size", ConfigUtil.kafkaBatch_size_config());
        kafkaProps.put("key.serializer", ConfigUtil.kafkaKey_serializer_class_config());
        kafkaProps.put("value.serializer", ConfigUtil.kafkaValue_serializer_class_config());

        kafkaProducer = new KafkaProducer<>(kafkaProps);
    }

    public void send(RowData rowData) {
        kafkaProducer.send(new ProducerRecord<>(ConfigUtil.kafkaTopic(), null, rowData));
    }
}

public class CanalClient {
	// ...
    private KafkaSender kafkaSender;
    
        // 开始监听
    public void start() {
        	...
            RowData rowData = new RowData(binlogMsgMap);
            kafkaSender.send(rowData);
    }

# 创建topic
bin/kafka-topics.sh --create --zookeeper node1:2181 --topic ods_itcast_shop_mysql --replication-factor 3 --partitions 3
# 创建控制台消费者测试
bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic ods_itcast_shop_mysql --from-beginning

#
#kafka的配置
#
# Kafka集群地址
bootstrap.servers="hadoop102:9092,hadoop103:9092,hadoop104:9092"
# ZooKeeper集群地址
zookeeper.connect="hadoop102:2181,hadoop103:2181,hadoop104:2181"
# 消费组ID
group.id="carlos"
# 自动提交拉取到消费端的消息offset到kafka
enable.auto.commit="true"
# 自动提交offset到zookeeper的时间间隔单位（毫秒）
auto.commit.interval.ms="5000"
# 每次消费最新的数据
auto.offset.reset="latest"
# kafka序列化器
key.serializer="org.apache.kafka.common.serialization.StringSerializer"
# kafka反序列化器
key.deserializer="org.apache.kafka.common.serialization.StringDeserializer"

# ip库本地文件路径
ip.file.path="E:/2021-05-10/carlos_shop_parent/data/qqwry.dat"

# Redis配置
redis.server.ip="hadoop102"
redis.server.port=6379

# MySQL配置
mysql.server.ip="hadoop102"
mysql.server.port=3306
mysql.server.database="itcast_shop"
mysql.server.username="root"
mysql.server.password="root"

# Kafka Topic 名称
input.topic.canal="ods_carlos_shop_mysql"
# Kafka click_log topic 名称
input.topic.click_log="ods_carlos_click_log"
# Kafka 购物车 topic 名称
input.topic.cart="ods_carlos_cart"
# kafka 评论 topic 名称
input.topic.comments="ods_carlos_comments"

# Druid Kafka 数据源 topic名称
output.topic.order="dwd_order"
output.topic.order_detail="dwd_order_detail"
output.topic.cart="dwd_cart"
output.topic.clicklog="dwd_click_log"
output.topic.goods="dwd_goods"
output.topic.ordertimeout="dwd_order_timeout"
output.topic.comments="dwd_comments"

# HBase订单明细表配置
hbase.table.orderdetail="dwd_order_detail"
hbase.table.family="detail"

log4j.rootLogger=warn,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

import com.typesafe.config.{Config, ConfigFactory}

object GlobalConfigUtil {
  private val config: Config = ConfigFactory.load()

  val `bootstrap.servers` = config.getString("bootstrap.servers")
  val `zookeeper.connect` = config.getString("zookeeper.connect")
  val `input.topic.canal` = config.getString("input.topic.canal")
  val `input.topic.click_log` = config.getString("input.topic.click_log")
  val `input.topic.comments` = config.getString("input.topic.comments")
  val `group.id` = config.getString("group.id")
  val `enable.auto.commit` = config.getString("enable.auto.commit")
  val `auto.commit.interval.ms` = config.getString("auto.commit.interval.ms")
  val `auto.offset.reset` = config.getString("auto.offset.reset")
  val `key.serializer` = config.getString("key.serializer")
  val `key.deserializer` = config.getString("key.deserializer")
  val `output.topic.order` = config.getString("output.topic.order")
  val `output.topic.order_detail` = config.getString("output.topic.order_detail")
  val `output.topic.cart` = config.getString("output.topic.cart")
  val `output.topic.clicklog` = config.getString("output.topic.clicklog")
  val `output.topic.goods` = config.getString("output.topic.goods")
  val `output.topic.ordertimeout` = config.getString("output.topic.ordertimeout")
  val `output.topic.comments` =  config.getString("output.topic.comments")
  val `hbase.table.orderdetail` = config.getString("hbase.table.orderdetail")
  val `hbase.table.family` = config.getString("hbase.table.family")
  val `redis.server.ip` = config.getString("redis.server.ip")
  val `redis.server.port`: String = config.getString("redis.server.port")
  val `ip.file.path` = config.getString("ip.file.path")
  val `mysql.server.ip` = config.getString("mysql.server.ip")
  val `mysql.server.port` = config.getString("mysql.server.port")
  val `mysql.server.database` = config.getString("mysql.server.database")
  val `mysql.server.username` = config.getString("mysql.server.username")
  val `mysql.server.password` = config.getString("mysql.server.password")
  val `input.topic.cart` = config.getString("input.topic.cart")

  def main(args: Array[String]): Unit = {
    println(`bootstrap.servers`)
    println(`zookeeper.connect`)
    println(`input.topic.canal`)
    println(`input.topic.click_log`)
    println(`group.id`)
    println(`enable.auto.commit`)
    println(`auto.commit.interval.ms`)
    println(`auto.offset.reset`)
    println(`output.topic.order`)
    println(`hbase.table.family`)
  }
}

import java.util.Properties

object KafkaProp {
  /**
    * 读取Kafka属性配置
    * @return
    */
  def getProperties() = {
    // 1. 读取Kafka配置
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", GlobalConfigUtil.`bootstrap.servers`)
    properties.setProperty("zookeeper.connect", GlobalConfigUtil.`zookeeper.connect`)
    properties.setProperty("group.id", GlobalConfigUtil.`group.id`)
    properties.setProperty("enable.auto.commit", GlobalConfigUtil.`enable.auto.commit`)
    properties.setProperty("auto.commit.interval.ms", GlobalConfigUtil.`auto.commit.interval.ms`)
    properties.setProperty("auto.offset.reset", GlobalConfigUtil.`auto.offset.reset`)
    properties.setProperty("key.serializer", GlobalConfigUtil.`key.serializer`)
    properties.setProperty("key.deserializer", GlobalConfigUtil.`key.deserializer`)

    properties
  }
}

val orderETL = new OrderETL(env)
orderETL.process()

# Kafka 购物车 topic名称
input.topic.cart="ods_carlos_cart"

val `input.topic.cart` = config.getString("input.topic.cart")

# kafka 评论 topic名称
input.topic.comments="ods_carlos_shop_comments"

val `input.topic.comments` = config.getString("input.topic.comments")

/**
  * 点击流处理逻辑
  */
class CommentsETL(env:StreamExecutionEnvironment) extends BaseETL[String] {

  /**
   * 业务处理接口
   */
  override def process(): Unit = {
    // 1. 整合Kafka
    val commentsDS: DataStream[String] = getKafkaDataStream(GlobalConfigUtil.`input.topic.comments`)
    commentsDS.print()
  }
}

# Kafka click_log topic名称
input.topic.click_log="ods_carlos_click_log"

val `input.topic.click_log` = config.getString("input.topic.click_log")

/**
  * 点击流处理逻辑
  */
class ClickLogETL(env:StreamExecutionEnvironment) extends BaseETL[String] {
  /**
    * 业务处理接口
    */
  override def process(): Unit = {
      // 1. 整合kafka
    val clickLogDS: DataStream[String] = getKafkaDataStream(GlobalConfigUtil.`input.topic.click_log`)
    clickLogDS.print()
  }
}

# Redis配置
redis.server.ip="hadoop102"
redis.server.port=6379

# MySQL配置
mysql.server.ip="hadoop102"
mysql.server.port=3306
mysql.server.database="itcast_shop"
mysql.server.username="root"
mysql.server.password="root"

val `redis.server.ip` = config.getString("redis.server.ip")
val `redis.server.port` = config.getString("redis.server.port")
val `mysql.server.ip` = config.getString("mysql.server.ip")
val `mysql.server.port` = config.getString("mysql.server.port")
val `mysql.server.database` = config.getString("mysql.server.database")
val `mysql.server.username` = config.getString("mysql.server.username")
val `mysql.server.password` = config.getString("mysql.server.password")

object RedisUtil {
  val config = new JedisPoolConfig()

  //是否启用后进先出, 默认true
  config.setLifo(true)
  //最大空闲连接数, 默认8个
  config.setMaxIdle(8)
  //最大连接数, 默认8个
  config.setMaxTotal(1000)
  //获取连接时的最大等待毫秒数(如果设置为阻塞时BlockWhenExhausted),如果超时就抛异常, 小于零:阻塞不确定的时间,  默认-1
  config.setMaxWaitMillis(-1)
  //逐出连接的最小空闲时间 默认1800000毫秒(30分钟)
  config.setMinEvictableIdleTimeMillis(1800000)
  //最小空闲连接数, 默认0
  config.setMinIdle(0)
  //每次逐出检查时 逐出的最大数目 如果为负数就是 : 1/abs(n), 默认3
  config.setNumTestsPerEvictionRun(3)
  //对象空闲多久后逐出, 当空闲时间>该值 且 空闲连接>最大空闲数 时直接逐出,不再根据MinEvictableIdleTimeMillis判断  (默认逐出策略)
  config.setSoftMinEvictableIdleTimeMillis(1800000)
  //在获取连接的时候检查有效性, 默认false
  config.setTestOnBorrow(false)
  //在空闲时检查有效性, 默认false
  config.setTestWhileIdle(false)

  var jedisPool: JedisPool = new JedisPool(config, GlobalConfigUtil.`redis.server.ip`, GlobalConfigUtil.`redis.server.port`.toInt)

  /**
    * 获取Redis连接
    * @return
    */
  def getResouce() = {
    jedisPool.getResource
  }
}

/**
  * Redis维度数据同步业务
  */
class SyncDimDataETL(env: StreamExecutionEnvironment) extends MySqlBaseETL(env) {
  /**
    * 业务处理接口
    */
  override def process(): Unit = {
    
  }
}

    val syncRedisDimDataETL = new SyncRedisDimDataETL(env)
    syncRedisDimDataETL.process()

<dependency>
    <groupId>nl.basjes.parse.httpdlog</groupId>
    <artifactId>httpdlog-parser</artifactId>
    <version>5.2</version>
</dependency>

 #log_format  main   '$remote_addr - $remote_user [$time_local] [$msec] 	
 						[$request_time] [$http_host] "$request" '
                    '$status $body_bytes_sent "$request_body" "$http_referer" '
                    '"$http_user_agent" $http_x_forwarded_for'

%u %h %l %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-Agent}i\" \"%{Cookie}i\" \"%{Addr}i\"

## Flink实时etl

### 点击流消息实时拉宽处理

#### 拉宽点击流消息处理

#### 点击流消息实时拉宽测试

### 订单数据实时处理

#### 操作步骤

1、在etl包下创建OrderETL类，实现MySqlBaseETL

2、过滤出来表名为itcast_orders，且事件类型为 "insert" 的binlog消息

3、将RowData数据流转换为 OrderDBEntity数据源

4、为了方便落地到 Kafka，再将OrderDBEntity 转换为JSON字符串

5、将转换后的json字符串写入到kafka的**dwd_order**topic中


### 订单明细数据实时拉宽处理

#### 操作步骤

1、在etl包下创建OrderGoodsEtl类，实现MySqlBaseETl

2、先过滤出表名为 itcast_order_goods，且事件类型为 "insert" 的binlog消息

3、为了方便落地到 Kafka，再将OrderGoodsWideEntity转换为JSON字符串

4、根据以下方式拉宽订单明细数据

* 根据商品id，从redis中商品维度Hash获取商品数据
* 根据店铺id，从redis中门店维护Hash获取门店数据
* 根据商品数据的三级分类id，从redis中的商品分类Hash获取三级分类数据
* 根据三级分类数据的parentid，从redis中的商品分类Hash获取二级分类数据
* 根据二级分类数据的parentid，从redis中的商品分类Hash获取一级分类数据
* 根据门店数据的areaId，从redis中的组织机构Hash获取城市机构数据
* 根据城市机构数据的parentId，从redis中的组织机构Hash获取省份机构数据

5、将拉宽后的订单明细数据写入到hbase数据库中


### 商品消息实时拉宽处理

#### 操作步骤

1、在etl包下创建GoodsETL类，实现MySqlBaseETL

2、先过滤出表名为 itcast_goods

3、根据以下方式拉宽订单明细数据

* 根据商品id，从redis中商品维度Hash获取商品数据
* 根据店铺id，从redis中门店维护Hash获取门店数据
* 根据商品数据的三级分类id，从redis中的商品分类Hash获取三级分类数据
* 根据三级分类数据的parentid，从redis中的商品分类Hash获取二级分类数据
* 根据二级分类数据的parentid，从redis中的商品分类Hash获取一级分类数据
* 根据门店数据的areaId，从redis中的组织机构Hash获取城市机构数据
* 根据城市机构数据的parentId，从redis中的组织机构Hash获取省份机构数据

4、为了方便落地到 Kafka，再将GoodsWideBean转换为JSON字符串

#### 注册IP库为Flink分布式缓存

```scala
env.registerCachedFile(GlobalConfigUtil.`ip.file.path`, "qqwry.dat")

    // 将JSON转换为实体类
    val cartBeanDS = kafkaDS.map(CartBean(_))

cd /opt/module/
wget https://static.imply.io/release/imply-3.0.4.tar.gz

将该 `imply安装包\imply-3.0.4.tar.gz` 安装包上传到 /exports/softwares

tar -xvzf imply-3.0.4.tar.gz -C ../module
cd ../module/imply-3.0.4

CREATE DATABASE `druid` DEFAULT CHARACTER SET utf8;
CREATE DATABASE `pivot` DEFAULT CHARACTER SET utf8;

druid.zk.service.host=hadoop102:2181,hadoop103:2181,hadoop104:2181

druid.metadata.storage.type=mysql
druid.metadata.storage.connector.connectURI=jdbc:mysql://hadoop102:3306/druid
druid.metadata.storage.connector.user=root
druid.metadata.storage.connector.password=root

stateStore:
  type: mysql
  location: mysql
  connection: 'mysql://root:root@hadoop102:3306/pivot'

#!/bin/bash
#1. 判断参数个数
if [ $# -lt 1 ]
then
  echo Not Enough Arguement!
  exit;
fi
#2. 遍历集群所有机器
for host in hadoop102 hadoop103 hadoop104
do
  echo ====================  $host  ====================
  #3. 遍历所有目录，挨个发送
  for file in $@
  do
    #4 判断文件是否存在
    if [ -e $file ]
    then
      #5. 获取父目录
      pdir=$(cd -P $(dirname $file); pwd)
      #6. 获取当前文件的名称
      fname=$(basename $file)
      ssh $host "mkdir -p $pdir"
      rsync -av $pdir/$fname $host:$pdir
    else
      echo $file does not exists!
    fi
  done
done

[carlos@hadoop102 ~]$ xsync imply-3.0.4

# DRUID
export DRUID_HOME=/opt/module/imply-3.0.4

# 使用外部 zk 而不使用imply自带zk启动overlord和coordinator
[carlos@hadoop102 imply-3.0.4]$ bin/supervise -c /opt/module/imply-3.0.4/conf/supervise/master-no-zk.conf

[carlos@hadoop103 imply-3.0.4]$ bin/supervise -c /opt/module/imply-3.0.4/conf/supervise/data.conf

[carlos@hadoop104 imply-3.0.4]$ bin/supervise -c /opt/module/imply-3.0.4/conf/supervise/query.conf