前提概要：之前我们已经实现了动态分流，即通过TableProcessFunction1类把维度数据和事实数据进行了分流处理，接下来就是把数据写入Hbase表和Kafka主题表中：
hbaseDS.addSink(new DimSink());
kafkaDS.addSink(kafkaSink);
此时的动态分流后的2种数据类型大致为：
在代码注释种我已经详尽地介绍了输出数据的情况和代码逻辑，接下来我会以代码编写逻辑思路为主要讲解路线，可以参考代码逻辑来看；
维度表和事实表动态写入
Hbase动态写入
怎么去编写这个DimSink类呢？首先我们需要通过Phoenix来与Hbase建立联系，之前我们已经通过循环遍历配置表来完成Hbase对应的建表操作了，所以此时我们已经可以通过Phoenix的Sql语法来完成数据插入操作了；
所以这里我们应该先获取到要插入的表名，然后整理成对应的Sql语句，具体句法类似upsert into 表空间.表名(列名…) values (值…)；
但有一点要注意一下：
if(jsonObj.getString("type").equals("update")){
DimUtil.deleteCached(tableName,dataJsonObj.getString("id"));
}
//这一段代码的作用是在后续Hbase表数据查询优化过程种起作用的，在这部分内容里面可以先忽略不管；
其他代码内容主要就是Phoenix的链接、Sql语句的生成以及upsert语句编写；
Kafka动态写入
回到之前我们写过的MyKafkaUtil的工具类中，当时我们的Producer对象是这么生成的：
//封装FlinkKafkaProducer
public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
//返回的就是kafka的producer对象；
return new FlinkKafkaProducer<String>(KAFKA_SERVER, topic, new SimpleStringSchema());
}
这段代码的topic对象在方法的参数中就已经确定了，但是我们在这里的kafka主题表不止一个，如果要把所有的kafka主题数据都写入进去，我们只能根据配置表中对应的kafka主题，有多少个主题就生成多少个Producer对象，但即使是这样，如果业务中有了新的事实数据类型，需要一个新的kafka主题，那我们这里还必须停止集群，重新添加新的Producer对象生成代码，这明显是不合理的；所以，这里我们采用另外一种创建方式：
//这种生成的FlinkKafkaProducer可以支持多个topic的情形；
public static <T> FlinkKafkaProducer<T> getKafkaSinkBySchema(KafkaSerializationSchema<T> kafkaSerializationSchema) {
Properties props = new Properties();
props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFKA_SERVER);
//设置生产数据的超时时间
props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000+"");
//可以往默认的topic发送，也可以指定topic；
//KafkaSerializationSchema 序列化的类型；
//props 生产者配置信息；
//FlinkKafkaProducer.Semantic 语义，指定精准一致性；
//这样写可以发往多个主题；
return new FlinkKafkaProducer<T>(DEFAULT_TOPIC, kafkaSerializationSchema, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
生成的FlinkKafkaProducer对象可以动态的选择kafka主题名，所以接下来我们拉看看这个Prodecer对象是如何实现动态写入kafka主题数据的：
FlinkKafkaProducer<JSONObject> kafkaSink = MyKafkaUtil.getKafkaSinkBySchema(
new KafkaSerializationSchema<JSONObject>() {
@Override
public void open(SerializationSchema.InitializationContext context) throws Exception {
System.out.println("kafka序列化");
}
//核心方法，每一条数据的序列化；
@Override
//这里的两个byte[], byte[]分别指代K,V；只不过这里没用K，只用了V；注意：K不是这里的topic，topic不传输；
//K（即Topic的作用是为了帮助分配到不同的主题中）
public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
String sinkTopic = jsonObj.getString("sink_table");
JSONObject dataJsonObj = jsonObj.getJSONObject("data");
//两个参数分别表示：主题名和数据值；
//序列化的意义就是把数据转化为字节数组，所以这里的value转化为Byte类型来做；主题找到位置即可，不用序列化；
return new ProducerRecord<>(sinkTopic,dataJsonObj.toString().getBytes());
}
}
);
//在ProducerRecord方法中，通过获取数据流中的sinkTable（记住此时数据流已经是主题数据了），来动态的传入sinkTopic；
到此为止，分流数据保存到Hbase和Kafka中的操作就全部完成了；
总结：
1 接收 Kafka 数据，转换过后再过滤空值数据
对 Maxwell 抓取数据进行 ETL，有用的部分保留，没用的过滤掉
2 实现动态分流功能（重难点）
由于MaxWell是把全部数据统一写入一个Topic中, 这样显然不利于日后的数据处理。所以需要把各个表拆开处理。但是由于每个表有不同的特点，有些表是维度表，有些表是事实表，有的表既是事实表在某种情况下也是维度表。
在实时计算中一般把维度数据写入存储容器，一般是方便通过主键查询的数据库比如HBase,Redis,MySQL 等。一般把事实数据写入流中，进行进一步处理，最终形成宽表。但是这里用Redis的话会对内存要求非常高，所以这里采用HBase来存储维度数据；
但是作为Flink实时计算任务，如何得知哪些表是维度表，哪些是事实表呢？而这些表又应该采集哪些字段呢？
我们可以将上面的内容放到某一个地方，集中配置。这样的配置不适合写在配置文件中，因为业务端随着需求变化每增加一张表，就要修改配置重启计算程序。所以这里需要一种动态配置方案，把这种配置长期保存起来，一旦配置有变化，实时计算可以自动感知。这种可以有两个方案实现：
? 一种是用Zookeeper存储，通过Watch感知数据变化。
? 另一种是用mysql数据库存储，周期性的同步。
这里选择第二种方案，主要是mysql对于配置数据初始化和维护管理，用sql都比较方便，虽然周期性操作时效性差一点，但是配置变化并不频繁。
//在当前项目中，选择的是通过自己来维护配置信息表，而在实际项目中往往是通过web页面的方式来维护；
//举例说明：实际业务发生了一个事件：数据库的品牌表中添加了一个新的品牌，相当于往base_trademark表中添加了一条记录，binlog会将这条记录保存下来，Maxwell会监听到binlog的变化，然后将发生变化的这条记录以json的形式发给kafka(ods_base_db_m);BaseDBApp从kafaka的od_base_db_m中读取到这条记录，然后从MySql的配置表获取相关信息：
--维度数据 发送到HBase
--事实数据 发送到kafka的dwd层
3.配置phoenix和hbase的相关内容，创建对应的命名空间；（具体见PDF文件）
4 把分好的流保存到对应表、主题中
业务数据保存到 Kafka 的主题中
维度数据保存到 Hbase 的表中
/*
数据大概样式：
{"database":"gmall0709","table":"order_info","type":"update","ts":1626008658,"xid":254,"xoffset":4651,
"data":{"id":28993,"consignee":"蒋莲真","consignee_tel":"13494023727","total_amount":21996.00,
"order_status":"1001","user_id":438,"payment_way":null,"delivery_address":"第17大街第39号楼1单元977门",
"order_comment":"描述178284","out_trade_no":"352538349474172","trade_body":"十月稻田 长粒香大米 东北大米 东北香米 5kg等10件商品",
"create_time":"2021-07-11 21:04:17","operate_time":"2021-07-11 21:04:18","expire_time":"2021-07-11 21:19:17",
"process_status":null,"tracking_no":null,"parent_order_id":null,"img_url":"http://img.gmall.com/646283.jpg",
"province_id":29,"activity_reduce_amount":0.00,"coupon_reduce_amount":0.00,"original_total_amount":21989.00,
"feight_fee":7.00,"feight_fee_reduce":null,"refundable_time":null},
"old":{"operate_time":null}}
有些数据不全
*/
//数据中携带了表格的信息和type的信息，根据此可以去配置表中找到自己属于维度数据还是事实数据；
//log测试：
开启zk.sh、kk.sh、maxwell.sh、当前应用程序；再去开启业务数据jar包：
/opt/module/rt_dblog $java -jar gmall2020-mock-db-2020-11-27.jar
//查看mysql中是否接收到数据；
//配置信息表测试：
//测试，是否能读到刚才创建的table_process表的数据信息；
//手动添加一条数据，然后运行com.chenxu.gmall.realtime.utils.MyKafkaUtil；
//配置phoenix和HBase的API测试；
//开启zk、kk、hdfs、hbase、phoenix、maxwell、本应用程序，记得提前在gmall0709_realtime数据库中手动插入一条配置文件信息；
//此时phoenix会自动建表GMALL0709_REALTIME
//手动修改gmall0709数据库中base_trademark表的数据，应该会给出类似下面的输出；
//维度>>>>:2> {"database":"gmall0709","xid":18593,"data":{"id":17},"commit":true,"sink_table":"DIM_BASE_TRADEMARK","type":"insert","table":"base_trademark","ts":1626230909}
//hbase中不会接收到具体数据；
//再把刚才配置表中的sink_type改为kafka，把sink_table改为first
//此时Map会重新读取这条新的配置信息，加入到内存中，数据也会输出到kafka中；
//重启应用程序，会出现类似下面的输出：
//事实>>>>:4> {"database":"gmall0709","xid":19366,"data":{"id":19},"commit":true,"sink_table":"first","type":"insert","table":"base_trademark","ts":1626231206}
//但是kafka中不会接收到数据；
//所以接下来要做的，就是把对应的维度数据再保存到hbase表中，把事实数据输出到kafka中；
//测试TODO6 newSink函数，维度数据是否会存入对应HBase表中；
//启动zk、kk、hdfs、hbase、maxwell、phoenix、应用程序；
//手动修改刚才的配置表信息，改回hbase和DIM_BASE_TRADEMARK；
//再去base_trademark表中添加一下数据；
//输出一条信息如下：
//维度>>>>:2> {"database":"gmall0709","xid":30932,"data":{"tm_name":"20","id":20},"commit":true,"sink_table":"DIM_BASE_TRADEMARK","type":"insert","table":"base_trademark","ts":1626240730}
//向Phoenix插入数据的SQL:upsert into GMALL0709_REALTIME.DIM_BASE_TRADEMARK(tm_name,id) values ('20','20')
//在phoenix中查询数据是否插入：
//select * from GMALL0709_REALTIME.DIM_BASE_TRADEMARK;
//测试
//启动 hdfs、zk、kafka、Maxwell、hbase,kafka的dwd_order_info主题消费
//向 gmall2021_realtime 数据库的 table_process 表中插入测试数据
//比如：对应数据为：order_info insert kafka dwd_order_info id,consignee,consignee_tel,total_count,order_status,user_id,payment_way,delivery_address,order_comment,out_trade_no,trade_body,create_time,operate_time,expire_time,process_status,tracking_no,parent_order_id,img_url,province_id,activity_reduce_amount,coupon_reduce_amount,original_total_amount,feight_fee,feight_fee_reduce,refundable_time id
//运行 idea 中的 BaseDBApp
//运行 rt_dblog 下的 jar 包，模拟生成数据
//查看控制台输出以及在配置表中配置的 kafka 主题名消费情况（kafka中消费的都是订单数据）
//此时还可以再增加base_trademark表中的数据，看看控制台会不会输出维度表的信息，再去phoenix中查看是否有数据：
//select * from GMALL0709_REALTIME.DIM_BASE_TRADEMARK;
流程图总结如下所示：（要注意，这里具体维度数据和事实数据怎么用还没有实现，只是做了一个简单的表测试）
