• 尚硅谷大数据项目《在线教育之实时数仓》笔记004

    视频地址：尚硅谷大数据项目《在线教育之实时数仓》_哔哩哔哩_bilibili

    目录

    第8章 数仓开发之DIM层

    P024

    P025

    P026

    P027

    P028

    P029

    P030

    第8章 数仓开发之DIM层

    P024

    1. package com.atguigu.edu.realtime.app.func;
    2.  
    3. import com.alibaba.druid.pool.DruidDataSource;
    4. import com.alibaba.druid.pool.DruidPooledConnection;
    5. import com.alibaba.fastjson.JSON;
    6. import com.alibaba.fastjson.JSONObject;
    7. import com.atguigu.edu.realtime.bean.DimTableProcess;
    8. import com.atguigu.edu.realtime.common.EduConfig;
    9. import com.atguigu.edu.realtime.util.DruidDSUtil;
    10. import com.atguigu.edu.realtime.util.PhoenixUtil;
    11. import org.apache.flink.api.common.state.BroadcastState;
    12. import org.apache.flink.api.common.state.MapStateDescriptor;
    13. import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
    14. import org.apache.flink.configuration.Configuration;
    15. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
    16. import org.apache.flink.util.Collector;
    17.  
    18. import java.sql.*;
    19. import java.util.*;
    20.  
    21. public class DimBroadcastProcessFunction extends BroadcastProcessFunction {
    22.     private MapStateDescriptor tableProcessState;
    23.  
    24.     // 初始化配置表数据
    25.     private HashMap configMap = new HashMap<>();
    26.  
    27.     public DimBroadcastProcessFunction(MapStateDescriptor tableProcessState) {
    28.         this.tableProcessState = tableProcessState;
    29.     }
    30.  
    31.     @Override
    32.     public void open(Configuration parameters) throws Exception {
    33.         super.open(parameters);
    34.  
    35.         Connection connection = DriverManager.getConnection("jdbc:mysql://node001:3306/edu_config?" +
    36.                 "user=root&password=123456&useUnicode=true&" +
    37.                 "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false"
    38.         );
    39.  
    40.         PreparedStatement preparedStatement = connection.prepareStatement("select * from edu_config.table_process");
    41.         ResultSet resultSet = preparedStatement.executeQuery();
    42.         ResultSetMetaData metaData = resultSet.getMetaData();
    43.         while (resultSet.next()) {
    44.             JSONObject jsonObject = new JSONObject();
    45.             for (int i = 1; i <= metaData.getColumnCount(); i++) {
    46.                 String columnName = metaData.getColumnName(i);
    47.                 String columnValue = resultSet.getString(i);
    48.                 jsonObject.put(columnName, columnValue);
    49.             }
    50.             DimTableProcess dimTableProcess = jsonObject.toJavaObject(DimTableProcess.class);
    51.             configMap.put(dimTableProcess.getSourceTable(), dimTableProcess);
    52.         }
    53.         resultSet.close();
    54.         preparedStatement.close();
    55.         connection.close();
    56.     }
    57.  
    58.     /**
    59.      * @param value flinkCDC直接输入的json
    60.      * @param ctx
    61.      * @param out
    62.      * @throws Exception
    63.      */
    64.     @Override
    65.     public void processBroadcastElement(String value, Context ctx, Collector out) throws Exception {
    66.         //TODO 1 获取配置表数据解析格式
    67.         JSONObject jsonObject = JSON.parseObject(value);
    68.         String type = jsonObject.getString("op");
    69.         BroadcastState tableConfigState = ctx.getBroadcastState(tableProcessState);
    70.         if ("d".equals(type)) {
    71.             // 从状态中删除对应的表格
    72.             DimTableProcess before = jsonObject.getObject("before", DimTableProcess.class);
    73.             tableConfigState.remove(before.getSourceTable());
    74.             // 从configMap中删除对应的表格
    75.             configMap.remove(before.getSourceTable());
    76.         } else {
    77.             DimTableProcess after = jsonObject.getObject("after", DimTableProcess.class);
    78.             //TODO 3 将数据写入到状态 广播出去
    79.             tableConfigState.put(after.getSourceTable(), after);
    80.             //TODO 2 检查phoenix中是否存在表 不存在创建
    81.             String sinkTable = after.getSinkTable();
    82.             String sinkColumns = after.getSinkColumns();
    83.             String sinkPk = after.getSinkPk();
    84.             String sinkExtend = after.getSinkExtend();
    85.             checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
    86.         }
    87.     }
    88.  
    89.     /**
    90.      * @param value kafka中maxwell生成的json数据
    91.      * @param ctx
    92.      * @param out
    93.      * @throws Exception
    94.      */
    95.     @Override
    96.     public void processElement(JSONObject value, ReadOnlyContext ctx, Collector out) throws Exception {
    97.         //TODO 1 获取广播的配置数据
    98.  
    99.         //TODO 2 过滤出需要的维度字段
    100.  
    101.         //TODO 3 补充输出字段
    102.     }
    103. }

    P025

    8.3.2 根据MySQL的配置表，动态进行分流

    6）创建连接池工具类

    1. package com.atguigu.edu.realtime.app.func;
    2.  
    3. import com.alibaba.druid.pool.DruidDataSource;
    4. import com.alibaba.druid.pool.DruidPooledConnection;
    5. import com.alibaba.fastjson.JSON;
    6. import com.alibaba.fastjson.JSONObject;
    7. import com.atguigu.edu.realtime.bean.DimTableProcess;
    8. import com.atguigu.edu.realtime.common.EduConfig;
    9. import com.atguigu.edu.realtime.util.DruidDSUtil;
    10. import com.atguigu.edu.realtime.util.PhoenixUtil;
    11. import org.apache.flink.api.common.state.BroadcastState;
    12. import org.apache.flink.api.common.state.MapStateDescriptor;
    13. import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
    14. import org.apache.flink.configuration.Configuration;
    15. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
    16. import org.apache.flink.util.Collector;
    17.  
    18. import java.sql.*;
    19. import java.util.*;
    20.  
    21. public class DimBroadcastProcessFunction extends BroadcastProcessFunction {
    22.     private MapStateDescriptor tableProcessState;
    23.  
    24.     // 初始化配置表数据
    25.     private HashMap configMap = new HashMap<>();
    26.  
    27.     public DimBroadcastProcessFunction(MapStateDescriptor tableProcessState) {
    28.         this.tableProcessState = tableProcessState;
    29.     }
    30.  
    31.     @Override
    32.     public void open(Configuration parameters) throws Exception {
    33.         super.open(parameters);
    34.  
    35.         Connection connection = DriverManager.getConnection("jdbc:mysql://node001:3306/edu_config?" +
    36.                 "user=root&password=123456&useUnicode=true&" +
    37.                 "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false"
    38.         );
    39.  
    40.         PreparedStatement preparedStatement = connection.prepareStatement("select * from edu_config.table_process");
    41.         ResultSet resultSet = preparedStatement.executeQuery();
    42.         ResultSetMetaData metaData = resultSet.getMetaData();
    43.         while (resultSet.next()) {
    44.             JSONObject jsonObject = new JSONObject();
    45.             for (int i = 1; i <= metaData.getColumnCount(); i++) {
    46.                 String columnName = metaData.getColumnName(i);
    47.                 String columnValue = resultSet.getString(i);
    48.                 jsonObject.put(columnName, columnValue);
    49.             }
    50.             DimTableProcess dimTableProcess = jsonObject.toJavaObject(DimTableProcess.class);
    51.             configMap.put(dimTableProcess.getSourceTable(), dimTableProcess);
    52.         }
    53.         resultSet.close();
    54.         preparedStatement.close();
    55.         connection.close();
    56.     }
    57.  
    58.     /**
    59.      * @param value flinkCDC直接输入的json
    60.      * @param ctx
    61.      * @param out
    62.      * @throws Exception
    63.      */
    64.     @Override
    65.     public void processBroadcastElement(String value, Context ctx, Collector out) throws Exception {
    66.         //TODO 1 获取配置表数据解析格式
    67.         JSONObject jsonObject = JSON.parseObject(value);
    68.         String type = jsonObject.getString("op");
    69.         BroadcastState tableConfigState = ctx.getBroadcastState(tableProcessState);
    70.         if ("d".equals(type)) {
    71.             // 从状态中删除对应的表格
    72.             DimTableProcess before = jsonObject.getObject("before", DimTableProcess.class);
    73.             tableConfigState.remove(before.getSourceTable());
    74.             // 从configMap中删除对应的表格
    75.             configMap.remove(before.getSourceTable());
    76.         } else {
    77.             DimTableProcess after = jsonObject.getObject("after", DimTableProcess.class);
    78.             //TODO 3 将数据写入到状态 广播出去
    79.             tableConfigState.put(after.getSourceTable(), after);
    80.             //TODO 2 检查phoenix中是否存在表 不存在创建
    81.             String sinkTable = after.getSinkTable();
    82.             String sinkColumns = after.getSinkColumns();
    83.             String sinkPk = after.getSinkPk();
    84.             String sinkExtend = after.getSinkExtend();
    85.             checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
    86.         }
    87.     }
    88.  
    89.     private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
    90.         // create table if not exists table (id string pk, name string...)
    91.         // 拼接建表语句的sql
    92.         StringBuilder sql = new StringBuilder();
    93.         sql.append("create table if not exists " + EduConfig.HBASE_SCHEMA + "." + sinkTable + "(\n");
    94.  
    95.         // 判断主键
    96.         // 如果主键为空,默认使用id
    97.         if (sinkPk == null) {
    98.             sinkPk = "";
    99.         }
    100.         if (sinkExtend == null) {
    101.             sinkExtend = "";
    102.         }
    103.  
    104.         // 遍历字段拼接建表语句
    105.         String[] split = sinkColumns.split(",");
    106.         for (int i = 0; i < split.length; i++) {
    107.             sql.append(split[i] + " varchar");
    108.             if (split[i].equals(sinkPk)) {
    109.                 sql.append(" primary key");
    110.             }
    111.             if (i < split.length - 1) {
    112.                 sql.append(",\n");
    113.             }
    114.         }
    115.         sql.append(") ");
    116.         sql.append(sinkExtend);
    117.  
    118.         PhoenixUtil.executeDDL(sql.toString());
    119.     }
    120.  
    121.     /**
    122.      * @param value kafka中maxwell生成的json数据
    123.      * @param ctx
    124.      * @param out
    125.      * @throws Exception
    126.      */
    127.     @Override
    128.     public void processElement(JSONObject value, ReadOnlyContext ctx, Collector out) throws Exception {
    129.         //TODO 1 获取广播的配置数据
    130.  
    131.         //TODO 2 过滤出需要的维度字段
    132.  
    133.         //TODO 3 补充输出字段
    134.     }
    135. }

    P026

    P027

    启动hadoop、zookeeper、kafka、hbase。p41

    1. [atguigu@node001 ~]$ myhadoop.sh start
    2.  ================ 启动 hadoop集群 ================
    3.  ---------------- 启动 hdfs ----------------
    4. Starting namenodes on [node001]
    5. Starting datanodes
    6. Starting secondary namenodes [node003]
    7.  --------------- 启动 yarn ---------------
    8. Starting resourcemanager
    9. Starting nodemanagers
    10.  --------------- 启动 historyserver ---------------
    11. [atguigu@node001 ~]$ zookeeper.sh start
    12. ---------- zookeeper node001 启动 ----------
    13. ZooKeeper JMX enabled by default
    14. Using config: /opt/module/zookeeper/zookeeper-3.5.7/bin/../conf/zoo.cfg
    15. Starting zookeeper ... STARTED
    16. ---------- zookeeper node002 启动 ----------
    17. ZooKeeper JMX enabled by default
    18. Using config: /opt/module/zookeeper/zookeeper-3.5.7/bin/../conf/zoo.cfg
    19. Starting zookeeper ... STARTED
    20. ---------- zookeeper node003 启动 ----------
    21. ZooKeeper JMX enabled by default
    22. Using config: /opt/module/zookeeper/zookeeper-3.5.7/bin/../conf/zoo.cfg
    23. Starting zookeeper ... STARTED
    24. [atguigu@node001 ~]$ 
    25. [atguigu@node001 ~]$ 
    26. [atguigu@node001 ~]$ kafka.sh start
    27. --------------- node001 Kafka 启动 ---------------
    28. --------------- node002 Kafka 启动 ---------------
    29. --------------- node003 Kafka 启动 ---------------
    30. [atguigu@node001 ~]$ 
    31. [atguigu@node001 ~]$ 
    32. [atguigu@node001 ~]$ start-hbase.sh
    33. SLF4J: Class path contains multiple SLF4J bindings.
    34. SLF4J: Found binding in [jar:file:/opt/module/hbase/hbase-2.0.5/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    35. SLF4J: Found binding in [jar:file:/opt/module/hadoop/hadoop-3.1.3/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    36. SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
    37. SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
    38. running master, logging to /opt/module/hbase/hbase-2.0.5/logs/hbase-atguigu-master-node001.out
    39. node002: running regionserver, logging to /opt/module/hbase/hbase-2.0.5/logs/hbase-atguigu-regionserver-node002.out
    40. node003: running regionserver, logging to /opt/module/hbase/hbase-2.0.5/logs/hbase-atguigu-regionserver-node003.out
    41. node001: running regionserver, logging to /opt/module/hbase/hbase-2.0.5/logs/hbase-atguigu-regionserver-node001.out
    42. node002: running master, logging to /opt/module/hbase/hbase-2.0.5/logs/hbase-atguigu-master-node002.out
    43. [atguigu@node001 ~]$ jpsall 
    44. ================ node001 ================
    45. 4880 HMaster
    46. 4615 Kafka
    47. 4183 QuorumPeerMain
    48. 3017 DataNode
    49. 2858 NameNode
    50. 5083 HRegionServer
    51. 3676 JobHistoryServer
    52. 3454 NodeManager
    53. 5406 Jps
    54. ================ node002 ================
    55. 2080 ResourceManager
    56. 4050 Jps
    57. 3397 Kafka
    58. 3574 HRegionServer
    59. 2966 QuorumPeerMain
    60. 3719 HMaster
    61. 1833 DataNode
    62. 2233 NodeManager
    63. ================ node003 ================
    64. 3265 Kafka
    65. 2833 QuorumPeerMain
    66. 3634 Jps
    67. 2067 SecondaryNameNode
    68. 2245 NodeManager
    69. 1941 DataNode
    70. 3430 HRegionServer
    71. [atguigu@node001 ~]$ cd /opt/module/hbase/apache-phoenix-5.0.0-HBase-2.0-bin/
    72. [atguigu@node001 apache-phoenix-5.0.0-HBase-2.0-bin]$ cd bin
    73. [atguigu@node001 bin]$ pwd
    74. /opt/module/hbase/apache-phoenix-5.0.0-HBase-2.0-bin/bin
    75. [atguigu@node001 bin]$ ./sqlline.py node001:2181
    76. Setting property: [incremental, false]
    77. Setting property: [isolation, TRANSACTION_READ_COMMITTED]
    78. issuing: !connect jdbc:phoenix:node001:2181 none none org.apache.phoenix.jdbc.PhoenixDriver
    79. Connecting to jdbc:phoenix:node001:2181
    80. SLF4J: Class path contains multiple SLF4J bindings.
    81. SLF4J: Found binding in [jar:file:/opt/module/hbase/apache-phoenix-5.0.0-HBase-2.0-bin/phoenix-5.0.0-HBase-2.0-client.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    82. SLF4J: Found binding in [jar:file:/opt/module/hadoop/hadoop-3.1.3/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    83. SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
    84. 23/10/26 20:07:57 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
    85. Connected to: Phoenix (version 5.0)
    86. Driver: PhoenixEmbeddedDriver (version 5.0)
    87. Autocommit status: true
    88. Transaction isolation: TRANSACTION_READ_COMMITTED
    89. Building list of tables and columns for tab-completion (set fastconnect to true to skip)...
    90. 133/133 (100%) Done
    91. Done
    92. sqlline version 1.2.0
    93. 0: jdbc:phoenix:node001:2181> create schema EDU_REALTIME;
    94. No rows affected (3.039 seconds)
    95. 0: jdbc:phoenix:node001:2181> !tables 
    96. +------------+--------------+-------------+---------------+----------+------------+----------------------------+-----------------+--------------+--------------+
    97. | TABLE_CAT  | TABLE_SCHEM  | TABLE_NAME  |  TABLE_TYPE   | REMARKS  | TYPE_NAME  | SELF_REFERENCING_COL_NAME  | REF_GENERATION  | INDEX_STATE  | IMMUTABLE_RO |
    98. +------------+--------------+-------------+---------------+----------+------------+----------------------------+-----------------+--------------+--------------+
    99. |            | SYSTEM       | CATALOG     | SYSTEM TABLE  |          |            |                            |                 |              | false        |
    100. |            | SYSTEM       | FUNCTION    | SYSTEM TABLE  |          |            |                            |                 |              | false        |
    101. |            | SYSTEM       | LOG         | SYSTEM TABLE  |          |            |                            |                 |              | true         |
    102. |            | SYSTEM       | SEQUENCE    | SYSTEM TABLE  |          |            |                            |                 |              | false        |
    103. |            | SYSTEM       | STATS       | SYSTEM TABLE  |          |            |                            |                 |              | false        |
    104. +------------+--------------+-------------+---------------+----------+------------+----------------------------+-----------------+--------------+--------------+
    105. 0: jdbc:phoenix:node001:2181> !tables 
    106. +------------+---------------+---------------------------+---------------+----------+------------+----------------------------+-----------------+--------------+
    107. | TABLE_CAT  |  TABLE_SCHEM  |        TABLE_NAME         |  TABLE_TYPE   | REMARKS  | TYPE_NAME  | SELF_REFERENCING_COL_NAME  | REF_GENERATION  | INDEX_STATE  |
    108. +------------+---------------+---------------------------+---------------+----------+------------+----------------------------+-----------------+--------------+
    109. |            | SYSTEM        | CATALOG                   | SYSTEM TABLE  |          |            |                            |                 |              |
    110. |            | SYSTEM        | FUNCTION                  | SYSTEM TABLE  |          |            |                            |                 |              |
    111. |            | SYSTEM        | LOG                       | SYSTEM TABLE  |          |            |                            |                 |              |
    112. |            | SYSTEM        | SEQUENCE                  | SYSTEM TABLE  |          |            |                            |                 |              |
    113. |            | SYSTEM        | STATS                     | SYSTEM TABLE  |          |            |                            |                 |              |
    114. |            | EDU_REALTIME  | DIM_BASE_CATEGORY_INFO    | TABLE         |          |            |                            |                 |              |
    115. |            | EDU_REALTIME  | DIM_BASE_PROVINCE         | TABLE         |          |            |                            |                 |              |
    116. |            | EDU_REALTIME  | DIM_BASE_SOURCE           | TABLE         |          |            |                            |                 |              |
    117. |            | EDU_REALTIME  | DIM_BASE_SUBJECT_INFO     | TABLE         |          |            |                            |                 |              |
    118. |            | EDU_REALTIME  | DIM_CHAPTER_INFO          | TABLE         |          |            |                            |                 |              |
    119. |            | EDU_REALTIME  | DIM_COURSE_INFO           | TABLE         |          |            |                            |                 |              |
    120. |            | EDU_REALTIME  | DIM_KNOWLEDGE_POINT       | TABLE         |          |            |                            |                 |              |
    121. |            | EDU_REALTIME  | DIM_TEST_PAPER            | TABLE         |          |            |                            |                 |              |
    122. |            | EDU_REALTIME  | DIM_TEST_PAPER_QUESTION   | TABLE         |          |            |                            |                 |              |
    123. |            | EDU_REALTIME  | DIM_TEST_POINT_QUESTION   | TABLE         |          |            |                            |                 |              |
    124. |            | EDU_REALTIME  | DIM_TEST_QUESTION_INFO    | TABLE         |          |            |                            |                 |              |
    125. |            | EDU_REALTIME  | DIM_TEST_QUESTION_OPTION  | TABLE         |          |            |                            |                 |              |
    126. |            | EDU_REALTIME  | DIM_USER_INFO             | TABLE         |          |            |                            |                 |              |
    127. |            | EDU_REALTIME  | DIM_VIDEO_INFO            | TABLE         |          |            |                            |                 |              |
    128. +------------+---------------+---------------------------+---------------+----------+------------+----------------------------+-----------------+--------------+
    129. 0: jdbc:phoenix:node001:2181>

    P028

    1. package com.atguigu.edu.realtime.app.func;
    2.  
    3. import com.alibaba.druid.pool.DruidDataSource;
    4. import com.alibaba.druid.pool.DruidPooledConnection;
    5. import com.alibaba.fastjson.JSON;
    6. import com.alibaba.fastjson.JSONObject;
    7. import com.atguigu.edu.realtime.bean.DimTableProcess;
    8. import com.atguigu.edu.realtime.common.EduConfig;
    9. import com.atguigu.edu.realtime.util.DruidDSUtil;
    10. import com.atguigu.edu.realtime.util.PhoenixUtil;
    11. import org.apache.flink.api.common.state.BroadcastState;
    12. import org.apache.flink.api.common.state.MapStateDescriptor;
    13. import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
    14. import org.apache.flink.configuration.Configuration;
    15. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
    16. import org.apache.flink.util.Collector;
    17.  
    18. import java.sql.*;
    19. import java.util.*;
    20.  
    21. public class DimBroadcastProcessFunction extends BroadcastProcessFunction {
    22.     private MapStateDescriptor tableProcessState;
    23.  
    24.     // 初始化配置表数据
    25.     private HashMap configMap = new HashMap<>();
    26.  
    27.     public DimBroadcastProcessFunction(MapStateDescriptor tableProcessState) {
    28.         this.tableProcessState = tableProcessState;
    29.     }
    30.  
    31.     @Override
    32.     public void open(Configuration parameters) throws Exception {
    33.         super.open(parameters);
    34.  
    35.         Connection connection = DriverManager.getConnection("jdbc:mysql://node001:3306/edu_config?" +
    36.                 "user=root&password=123456&useUnicode=true&" +
    37.                 "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false"
    38.         );
    39.  
    40.         PreparedStatement preparedStatement = connection.prepareStatement("select * from edu_config.table_process");
    41.         ResultSet resultSet = preparedStatement.executeQuery();
    42.         ResultSetMetaData metaData = resultSet.getMetaData();
    43.         while (resultSet.next()) {
    44.             JSONObject jsonObject = new JSONObject();
    45.             for (int i = 1; i <= metaData.getColumnCount(); i++) {
    46.                 String columnName = metaData.getColumnName(i);
    47.                 String columnValue = resultSet.getString(i);
    48.                 jsonObject.put(columnName, columnValue);
    49.             }
    50.             DimTableProcess dimTableProcess = jsonObject.toJavaObject(DimTableProcess.class);
    51.             configMap.put(dimTableProcess.getSourceTable(), dimTableProcess);
    52.         }
    53.         resultSet.close();
    54.         preparedStatement.close();
    55.         connection.close();
    56.     }
    57.  
    58.     /**
    59.      * @param value flinkCDC直接输入的json
    60.      * @param ctx
    61.      * @param out
    62.      * @throws Exception
    63.      */
    64.     @Override
    65.     public void processBroadcastElement(String value, Context ctx, Collector out) throws Exception {
    66.         //TODO 1 获取配置表数据解析格式
    67.         JSONObject jsonObject = JSON.parseObject(value);
    68.         String type = jsonObject.getString("op");
    69.         BroadcastState tableConfigState = ctx.getBroadcastState(tableProcessState);
    70.         if ("d".equals(type)) {
    71.             // 从状态中删除对应的表格
    72.             DimTableProcess before = jsonObject.getObject("before", DimTableProcess.class);
    73.             tableConfigState.remove(before.getSourceTable());
    74.             // 从configMap中删除对应的表格
    75.             configMap.remove(before.getSourceTable());
    76.         } else {
    77.             DimTableProcess after = jsonObject.getObject("after", DimTableProcess.class);
    78.             //TODO 3 将数据写入到状态 广播出去
    79.             tableConfigState.put(after.getSourceTable(), after);
    80.             //TODO 2 检查phoenix中是否存在表 不存在创建
    81.             String sinkTable = after.getSinkTable();
    82.             String sinkColumns = after.getSinkColumns();
    83.             String sinkPk = after.getSinkPk();
    84.             String sinkExtend = after.getSinkExtend();
    85.             checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
    86.         }
    87.     }
    88.  
    89.     private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
    90.         // create table if not exists table (id string pk, name string...)
    91.         // 拼接建表语句的sql
    92.         StringBuilder sql = new StringBuilder();
    93.         sql.append("create table if not exists " + EduConfig.HBASE_SCHEMA + "." + sinkTable + "(\n");
    94.  
    95.         // 判断主键
    96.         // 如果主键为空,默认使用id
    97.         if (sinkPk == null) {
    98.             sinkPk = "";
    99.         }
    100.         if (sinkExtend == null) {
    101.             sinkExtend = "";
    102.         }
    103.  
    104.         // 遍历字段拼接建表语句
    105.         String[] split = sinkColumns.split(",");
    106.         for (int i = 0; i < split.length; i++) {
    107.             sql.append(split[i] + " varchar");
    108.             if (split[i].equals(sinkPk)) {
    109.                 sql.append(" primary key");
    110.             }
    111.             if (i < split.length - 1) {
    112.                 sql.append(",\n");
    113.             }
    114.         }
    115.         sql.append(") ");
    116.         sql.append(sinkExtend);
    117.  
    118.         PhoenixUtil.executeDDL(sql.toString());
    119.     }
    120.  
    121.     /**
    122.      * @param value kafka中maxwell生成的json数据
    123.      * @param ctx
    124.      * @param out
    125.      * @throws Exception
    126.      */
    127.     @Override
    128.     public void processElement(JSONObject value, ReadOnlyContext ctx, Collector out) throws Exception {
    129.         //TODO 1 获取广播的配置数据
    130.         ReadOnlyBroadcastState tableConfigState = ctx.getBroadcastState(tableProcessState);
    131.         DimTableProcess tableProcess = tableConfigState.get(value.getString("table"));
    132.         // 补充情况,防止kafka数据到的过快  造成数据丢失
    133.         if (tableProcess == null) {
    134.             tableProcess = configMap.get(value.getString("table"));
    135.         }
    136.         if (tableProcess != null) {
    137.             String type = value.getString("type");
    138.             if (type == null) {
    139.                 System.out.println("maxwell采集的数据不完整...");
    140.             } else {
    141.                 JSONObject data = value.getJSONObject("data");
    142.                 //TODO 2 过滤出需要的维度字段
    143.                 String sinkColumns = tableProcess.getSinkColumns();
    144.                 filterColumns(data, sinkColumns);
    145.                 //TODO 3 补充输出字段
    146.                 data.put("sink_table", tableProcess.getSinkTable());
    147.                 // 添加数据的类型
    148.                 data.put("type", type);
    149.                 out.collect(data);
    150.             }
    151.         }
    152.     }
    153.  
    154.     private void filterColumns(JSONObject data, String sinkColumns) {
    155.         Set> entries = data.entrySet();
    156.         List stringList = Arrays.asList(sinkColumns.split(","));
    157.         entries.removeIf(entry -> !stringList.contains(entry.getKey()));
    158.     }
    159. }

    P029

    1. package com.atguigu.edu.realtime.util;
    2.  
    3. import com.alibaba.druid.pool.DruidDataSource;
    4. import com.alibaba.druid.pool.DruidPooledConnection;
    5. import com.alibaba.fastjson.JSONObject;
    6. import com.atguigu.edu.realtime.common.EduConfig;
    7. import org.apache.commons.beanutils.BeanUtils;
    8. import org.apache.commons.lang3.StringUtils;
    9.  
    10. import java.lang.reflect.InvocationTargetException;
    11. import java.sql.PreparedStatement;
    12. import java.sql.ResultSet;
    13. import java.sql.ResultSetMetaData;
    14. import java.sql.SQLException;
    15. import java.util.ArrayList;
    16. import java.util.List;
    17. import java.util.Map;
    18. import java.util.Set;
    19.  
    20. public class PhoenixUtil {
    21.     private static DruidDataSource druidDataSource = DruidDSUtil.getDruidDataSource();
    22.  
    23.     public static void executeDDL(String sqlString) {
    24.         DruidPooledConnection connection = null;
    25.         PreparedStatement preparedStatement = null;
    26.         try {
    27.             connection = druidDataSource.getConnection();
    28.         } catch (SQLException throwables) {
    29.             throwables.printStackTrace();
    30.             System.out.println("连接池获取连接异常...");
    31.         }
    32.  
    33.         try {
    34.             preparedStatement = connection.prepareStatement(sqlString);
    35.         } catch (SQLException throwables) {
    36.             throwables.printStackTrace();
    37.             System.out.println("编译sql异常...");
    38.         }
    39.  
    40.         try {
    41.             preparedStatement.execute();
    42.         } catch (SQLException throwables) {
    43.             throwables.printStackTrace();
    44.             System.out.println("建表语句错误...");
    45.         }
    46.  
    47.         // 关闭资源
    48.         try {
    49.             preparedStatement.close();
    50.         } catch (SQLException throwables) {
    51.             throwables.printStackTrace();
    52.         }
    53.  
    54.         try {
    55.             connection.close();
    56.         } catch (SQLException throwables) {
    57.             throwables.printStackTrace();
    58.         }
    59.     }
    60.  
    61.     public static void executeDML(String sinkTable, JSONObject jsonObject) {
    62.         // TODO 2 拼接sql语言
    63.         StringBuilder sql = new StringBuilder();
    64.         Set> entries = jsonObject.entrySet();
    65.         ArrayList columns = new ArrayList<>();
    66.         ArrayList values = new ArrayList<>();
    67.         StringBuilder symbols = new StringBuilder();
    68.         for (Map.Entry entry : entries) {
    69.             columns.add(entry.getKey());
    70.             values.add(entry.getValue());
    71.             symbols.append("?,");
    72.         }
    73.  
    74.         sql.append("upsert into " + EduConfig.HBASE_SCHEMA + "." + sinkTable + "(");
    75.  
    76.         // 拼接列名
    77.         String columnsStrings = StringUtils.join(columns, ",");
    78.         String symbolStr = symbols.substring(0, symbols.length() - 1).toString();
    79.         sql.append(columnsStrings)
    80.                 .append(")values(")
    81.                 .append(symbolStr)
    82.                 .append(")");
    83.  
    84.         DruidPooledConnection connection = null;
    85.         try {
    86.             connection = druidDataSource.getConnection();
    87.         } catch (SQLException throwables) {
    88.             throwables.printStackTrace();
    89.             System.out.println("连接池获取连接异常...");
    90.         }
    91.  
    92.         PreparedStatement preparedStatement = null;
    93.         try {
    94.             preparedStatement = connection.prepareStatement(sql.toString());
    95.             // 传入参数
    96.             for (int i = 0; i < values.size(); i++) {
    97.                 preparedStatement.setObject(i + 1, values.get(i) + "");
    98.             }
    99.         } catch (SQLException throwables) {
    100.             throwables.printStackTrace();
    101.             System.out.println("编译sql异常...");
    102.         }
    103.  
    104.         try {
    105.             preparedStatement.executeUpdate();
    106.         } catch (SQLException throwables) {
    107.             throwables.printStackTrace();
    108.             System.out.println("写入phoenix错误...");
    109.         }
    110.  
    111.         try {
    112.             preparedStatement.close();
    113.         } catch (SQLException throwables) {
    114.             throwables.printStackTrace();
    115.         }
    116.  
    117.         try {
    118.             connection.close();
    119.         } catch (SQLException throwables) {
    120.             throwables.printStackTrace();
    121.         }
    122.     }
    123.  
    124.     public static  List queryList(String sql, Class clazz) {
    125.         ArrayList resultList = new ArrayList<>();
    126.         DruidPooledConnection connection = null;
    127.         PreparedStatement preparedStatement = null;
    128.         try {
    129.             connection = druidDataSource.getConnection();
    130.             preparedStatement = connection.prepareStatement(sql);
    131.             ResultSet resultSet = preparedStatement.executeQuery();
    132.             ResultSetMetaData metaData = resultSet.getMetaData();
    133.             while (resultSet.next()) {
    134.                 T obj = clazz.newInstance();
    135.                 for (int i = 1; i <= metaData.getColumnCount(); i++) {
    136.                     String columnName = metaData.getColumnName(i);
    137.                     Object columnValue = resultSet.getObject(i);
    138.                     BeanUtils.setProperty(obj, columnName, columnValue);
    139.                 }
    140.                 resultList.add(obj);
    141.             }
    142.         } catch (Exception e) {
    143.             e.printStackTrace();
    144.         }
    145.  
    146.         if (preparedStatement != null) {
    147.             try {
    148.                 preparedStatement.close();
    149.             } catch (SQLException throwables) {
    150.                 throwables.printStackTrace();
    151.             }
    152.         }
    153.  
    154.         if (connection != null) {
    155.             try {
    156.                 connection.close();
    157.             } catch (SQLException throwables) {
    158.                 throwables.printStackTrace();
    159.             }
    160.         }
    161.         return resultList;
    162.     }
    163. }
      164.  

      1. package com.atguigu.edu.realtime.app.func;
      2.  
      3. import com.alibaba.fastjson.JSONObject;
      4. import com.atguigu.edu.realtime.util.DimUtil;
      5. import com.atguigu.edu.realtime.util.PhoenixUtil;
      6. import org.apache.flink.streaming.api.functions.sink.SinkFunction;
      7.  
      8. public class DimPhoenixSinkFunc implements SinkFunction {
      9.     @Override
      10.     public void invoke(JSONObject jsonObject, Context context) throws Exception {
      11.         // TODO 1 获取输出的表名
      12.         String sinkTable = jsonObject.getString("sink_table");
      13.  
      14. //        String type = jsonObject.getString("type");
      15. //        String id = jsonObject.getString("id");
      16.         jsonObject.remove("sink_table");
      17. //        jsonObject.remove("type");
      18.  
      19.         // TODO 2 使用工具类写出数据
      20.         PhoenixUtil.executeDML(sinkTable, jsonObject);
      21.  
      22.         // TODO 3 如果类型为update，删除redis对应缓存
      23. //        if ("update".equals(type)) {
      24. //            DimUtil.deleteCached(sinkTable, id);
      25. //        }
      26.     }
      27. }
       

      P030
       

       
 
      启动hadoop、zookeeper、kafka、hbase、maxwell、apache-phoenix-5.0.0-HBase。
       

       

      1. org.apache.phoenix.schema.ColumnNotFoundException: ERROR 504 (42703): Undefined column. columnName=EDU_REALTIME.DIM_BASE_PROVINCE.TYPE
      2. 	at org.apache.phoenix.schema.PTableImpl.getColumnForColumnName(PTableImpl.java:828)
      3. 	at org.apache.phoenix.compile.FromCompiler$SingleTableColumnResolver.resolveColumn(FromCompiler.java:477)
      4. 	at org.apache.phoenix.compile.UpsertCompiler.compile(UpsertCompiler.java:452)
      5. 	at org.apache.phoenix.jdbc.PhoenixStatement$ExecutableUpsertStatement.compilePlan(PhoenixStatement.java:784)
      6. 	at org.apache.phoenix.jdbc.PhoenixStatement$ExecutableUpsertStatement.compilePlan(PhoenixStatement.java:770)
      7. 	at org.apache.phoenix.jdbc.PhoenixStatement$2.call(PhoenixStatement.java:401)
      8. 	at org.apache.phoenix.jdbc.PhoenixStatement$2.call(PhoenixStatement.java:391)
      9. 	at org.apache.phoenix.call.CallRunner.run(CallRunner.java:53)
      10. 	at org.apache.phoenix.jdbc.PhoenixStatement.executeMutation(PhoenixStatement.java:390)
      11. 	at org.apache.phoenix.jdbc.PhoenixStatement.executeMutation(PhoenixStatement.java:378)
      12. 	at org.apache.phoenix.jdbc.PhoenixPreparedStatement.executeUpdate(PhoenixPreparedStatement.java:206)
      13. 	at com.alibaba.druid.pool.DruidPooledPreparedStatement.executeUpdate(DruidPooledPreparedStatement.java:255)
      14. 	at com.atguigu.edu.realtime.util.PhoenixUtil.executeDML(PhoenixUtil.java:105)
      15. 	at com.atguigu.edu.realtime.app.func.DimPhoenixSinkFunc.invoke(DimPhoenixSinkFunc.java:19)
      16. 	at com.atguigu.edu.realtime.app.func.DimPhoenixSinkFunc.invoke(DimPhoenixSinkFunc.java:7)
      17. 	at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54)
      18. 	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
      19. 	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
      20. 	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
      21. 	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
      22. 	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
      23. 	at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
      24. 	at com.atguigu.edu.realtime.app.func.DimBroadcastProcessFunction.processElement(DimBroadcastProcessFunction.java:148)
      25. 	at com.atguigu.edu.realtime.app.func.DimBroadcastProcessFunction.processElement(DimBroadcastProcessFunction.java:21)
      26. 	at org.apache.flink.streaming.api.operators.co.CoBroadcastWithNonKeyedOperator.processElement1(CoBroadcastWithNonKeyedOperator.java:110)
      27. 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.processRecord1(StreamTwoInputProcessorFactory.java:217)
      28. 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.lambda$create$0(StreamTwoInputProcessorFactory.java:183)
      29. 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory$StreamTaskNetworkOutput.emitRecord(StreamTwoInputProcessorFactory.java:266)
      30. 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
      31. 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
      32. 	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
      33. 	at org.apache.flink.streaming.runtime.io.StreamMultipleInputProcessor.processInput(StreamMultipleInputProcessor.java:85)
      34. 	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:542)
      35. 	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
      36. 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
      37. 	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
      38. 	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
      39. 	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
      40. 	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
      41. 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
      42. 	at java.lang.Thread.run(Thread.java:748)
      43. 写入phoenix错误...
       

       
 
      直接使用最终的项目代码运行，懒得改bug了。。。
       

       

      1. [atguigu@node001 ~]$ cd /opt/module/hbase/apache-phoenix-5.0.0-HBase-2.0-bin/bin/
      2. [atguigu@node001 bin]$ ./sqlline.py node001:2181
      3. Setting property: [incremental, false]
      4. Setting property: [isolation, TRANSACTION_READ_COMMITTED]
      5. issuing: !connect jdbc:phoenix:node001:2181 none none org.apache.phoenix.jdbc.PhoenixDriver
      6. Connecting to jdbc:phoenix:node001:2181
      7. SLF4J: Class path contains multiple SLF4J bindings.
      8. SLF4J: Found binding in [jar:file:/opt/module/hbase/apache-phoenix-5.0.0-HBase-2.0-bin/phoenix-5.0.0-HBase-2.0-client.jar!/org/slf4j/impl/StaticLoggerBinder.class]
      9. SLF4J: Found binding in [jar:file:/opt/module/hadoop/hadoop-3.1.3/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
      10. SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
      11. 23/10/31 11:04:03 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
      12. Connected to: Phoenix (version 5.0)
      13. Driver: PhoenixEmbeddedDriver (version 5.0)
      14. Autocommit status: true
      15. Transaction isolation: TRANSACTION_READ_COMMITTED
      16. Building list of tables and columns for tab-completion (set fastconnect to true to skip)...
      17. 246/246 (100%) Done
      18. Done
      19. sqlline version 1.2.0
      20. 0: jdbc:phoenix:node001:2181> !tables 
      21. +------------+---------------+---------------------------+---------------+----------+------------+----------------------------+-----------------+--------------+
      22. | TABLE_CAT  |  TABLE_SCHEM  |        TABLE_NAME         |  TABLE_TYPE   | REMARKS  | TYPE_NAME  | SELF_REFERENCING_COL_NAME  | REF_GENERATION  | INDEX_STATE  |
      23. +------------+---------------+---------------------------+---------------+----------+------------+----------------------------+-----------------+--------------+
      24. |            | SYSTEM        | CATALOG                   | SYSTEM TABLE  |          |            |                            |                 |              |
      25. |            | SYSTEM        | FUNCTION                  | SYSTEM TABLE  |          |            |                            |                 |              |
      26. |            | SYSTEM        | LOG                       | SYSTEM TABLE  |          |            |                            |                 |              |
      27. |            | SYSTEM        | SEQUENCE                  | SYSTEM TABLE  |          |            |                            |                 |              |
      28. |            | SYSTEM        | STATS                     | SYSTEM TABLE  |          |            |                            |                 |              |
      29. |            | EDU_REALTIME  | DIM_BASE_CATEGORY_INFO    | TABLE         |          |            |                            |                 |              |
      30. |            | EDU_REALTIME  | DIM_BASE_PROVINCE         | TABLE         |          |            |                            |                 |              |
      31. |            | EDU_REALTIME  | DIM_BASE_SOURCE           | TABLE         |          |            |                            |                 |              |
      32. |            | EDU_REALTIME  | DIM_BASE_SUBJECT_INFO     | TABLE         |          |            |                            |                 |              |
      33. |            | EDU_REALTIME  | DIM_CHAPTER_INFO          | TABLE         |          |            |                            |                 |              |
      34. |            | EDU_REALTIME  | DIM_COURSE_INFO           | TABLE         |          |            |                            |                 |              |
      35. |            | EDU_REALTIME  | DIM_KNOWLEDGE_POINT       | TABLE         |          |            |                            |                 |              |
      36. |            | EDU_REALTIME  | DIM_TEST_PAPER            | TABLE         |          |            |                            |                 |              |
      37. |            | EDU_REALTIME  | DIM_TEST_PAPER_QUESTION   | TABLE         |          |            |                            |                 |              |
      38. |            | EDU_REALTIME  | DIM_TEST_POINT_QUESTION   | TABLE         |          |            |                            |                 |              |
      39. |            | EDU_REALTIME  | DIM_TEST_QUESTION_INFO    | TABLE         |          |            |                            |                 |              |
      40. |            | EDU_REALTIME  | DIM_TEST_QUESTION_OPTION  | TABLE         |          |            |                            |                 |              |
      41. |            | EDU_REALTIME  | DIM_USER_INFO             | TABLE         |          |            |                            |                 |              |
      42. |            | EDU_REALTIME  | DIM_VIDEO_INFO            | TABLE         |          |            |                            |                 |              |
      43. +------------+---------------+---------------------------+---------------+----------+------------+----------------------------+-----------------+--------------+
      44. 0: jdbc:phoenix:node001:2181> select * from EDU_REALTIME.DIM_BASE_CATEGORY_INFO;
      45. +-----+----------------+----------------------+--------------+----------+
      46. | ID  | CATEGORY_NAME  |     CREATE_TIME      | UPDATE_TIME  | DELETED  |
      47. +-----+----------------+----------------------+--------------+----------+
      48. | 1   | 编程技术           | 2021-09-24 22:19:37  | null         | 0        |
      49. +-----+----------------+----------------------+--------------+----------+
      50. 1 row selected (0.333 seconds)
      51. 0: jdbc:phoenix:node001:2181> 
       

       
 
      😘👌💕
       

      
                
                    
                

                
    164. 

                    相关阅读:
      
                    
L2 数据仓库和Hive环境配置                            
      
【Hack The Box】windows练习-- blue                            
      
03 最小CMake项目                            
      
【java学习—十五】创建多线程的两种方式（2）                            
      
中国财政科学研究院党委书记、院长刘尚希一行莅临麒麟信安调研                            
      
已解决ModuleNotFoundError: No module named‘ pip‘（重新安装pip的两种方式）                            
      
MySQL递归查询所有子集                            
      
Endpoint Central自动化软件部署                            
      
Tailscale的子网路由和出口节点                            
      
GUI编程--PyQt5--控件                            
      
                          
                
      165. 
                
    165. 
                    原文地址：https://blog.csdn.net/weixin_44949135/article/details/134060188
                
      166.