点一下关注吧非常感谢持续更新----------------------Java篇开始了---------* MyBatis 更新完毕* 目前开始更新 Spring一起深入浅出目前已经更新到了---------* Hadoop已更完* HDFS已更完* MapReduce已更完* Hive已更完* Flume已更完* Sqoop已更完* Zookeeper已更完* HBase已更完* Redis 已更完* Kafka已更完* Spark已更完* Flink已更完* ClickHouse已更完* Kudu已更完* Druid已更完* Kylin已更完* Elasticsearch已更完* DataX已更完* Tez已更完* 数据挖掘已更完* Prometheus已更完* Grafana已更完* 离线数仓已更完* 实时数仓正在更新…章节内容----* ODS层* 将 Kafka 中的维度表写入 DIMDIM 层处理-------### 实现思路最原始的表 MySQL 中 area 到 HBase 中 转换 area 表 到 地区ID、地区的名字、城市ID、城市的名字、省份 ID、省份的名字 到 HBase 中在分析交易过程时可以通过卖家、买家、商品和时间等维度描述交易发生的环境所以维度的作用一般是查询约束、分类汇总以及排序等。### HBaseReader class HBaseReader extends RichSourceFunction[(String, String)] { private var conn: Connection null private var table: Table null private var scan: Scan null override def open(parameters: configuration.Configuration): Unit { val conf: Configuration HBaseConfiguration.create() conf.set(HConstants.ZOOKEEPER_QUORUM, “h121.wzk.icu,h122.wzk.icu,h123.wzk.icu”) conf.set(HConstants.ZOOKEEPER_CLIENT_PORT, “2181”) conf.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 30000) conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 30000) val tableName: TableName TableName.valueOf(“wzk_area”) val cf1: String “f1” conn ConnectionFactory.createConnection(conf) table conn.getTable(tableName) scan new Scan() scan.addFamily(Bytes.toBytes(cf1)) } override def run(ctx: SourceFunction.SourceContext[(String, String)]): Unit { val rs: ResultScanner table.getScanner(scan) val iterator: util.Iterator[Result] rs.iterator() while (iterator.hasNext) { val result: Result iterator.next() val rowKey: String Bytes.toString(result.getRow) val buffer: StringBuffer new StringBuffer() for (cell: Cell - result.listCells().asScala) { val value: String Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength) buffer.append(value).append(“-”) } val valueString: String buffer.replace(buffer.length() - 1, buffer.length(), “”).toString ctx.collect((rowKey, valueString)) } } override def cancel(): Unit { } override def close(): Unit { try { if (table ! null) { table.close() } if (conn ! null) { conn.close() } } catch { case e: Exception println(e.getMessage) } } } ### HBaseWriterSink class HBaseWriterSink extends RichSinkFunction[String] { var connection: Connection _ var hbTable: Table _ override def open(parameters: Configuration): Unit { connection new ConnHBase().connToHbase hbTable connection.getTable(TableName.valueOf(“dim_wzk_area”)) } override def close(): Unit { if (hbTable ! null) { hbTable.close() } if (connection ! null) { connection.close() } } override def invoke(value: String, context: SinkFunction.Context[_]): Unit { insertDimArea(hbTable, value) } def insertDimArea(hbTable: Table, value: String): Unit { val infos: Array[String] value.split(“,”) val areaId: String infos(0).trim val aname: String infos(1).trim val cid: String infos(2).trim val city: String infos(3).trim val proid: String infos(4).trim val province: String infos(5).trim val put new Put(areaId.getBytes()) put.addColumn(“f1”.getBytes(), “aname”.getBytes(), aname.getBytes()) put.addColumn(“f1”.getBytes(), “cid”.getBytes(), cid.getBytes()) put.addColumn(“f1”.getBytes(), “city”.getBytes(), city.getBytes()) put.addColumn(“f1”.getBytes(), “proId”.getBytes(), proid.getBytes()) put.addColumn(“f1”.getBytes(), “province”.getBytes(), province.getBytes()) hbTable.put(put) } } ### AreaDetail case class AreaDetail(id: Int, name:String, pid: Int) ### AreaDetailInfo object AreaDetailInfo { def main(args: Array[String]): Unit { val env: StreamExecutionEnvironment StreamExecutionEnvironment.getExecutionEnvironment env.enableCheckpointing(5000) env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) val data: DataStream[(String, String)] env.addSource(new HBaseReader) val dataStream: DataStream[AreaDetail] data.map(x { val id: Int x._1.toInt val datas: Array[String] x._2.split(“-”) val name: String datas(5).trim val pid: Int datas(6).trim.toInt AreaDetail(id, name, pid) }) val tableEnv: StreamTableEnvironment StreamTableEnvironment.create(env) tableEnv.createTemporaryView(“wzk_area”,dataStream) val sql : String “” |select a.id as areaid,a.name as aname,a.pid as cid,b.name as city, c.id as proid,c.name as province |from wzk_area as a |inner join wzk_area as b on a.pid b.id |inner join wzk_area as c on b.pid c.id |“”“.stripMargin val areaTable: Table tableEnv.sqlQuery(sql) println(”— sqlQuery —“) val resultStream: DataStream[String] tableEnv.toRetractStreamRow.map(x { val row: Row x._2 val areaId: String row.getField(0).toString val aname: String row.getField(1).toString val cid: String row.getField(2).toString val city: String row.getField(3).toString val proid: String row.getField(4).toString val province: String row.getField(5).toString areaId “,” aname “,” cid “,” city “,” proid “,” province }) resultStream.addSink(new HBaseWriterSink) env.execute() } } ### DimArea case class DimArea(areaId:Int, aname:String, cid:Int, city:String, proId:Int, province:String) ### 运行测试我们需要运行 AreaDetailInfo.scala执行效果如下所示DW 层处理------DWData WareHouse 数据仓库层包含 DWD、DWS、DIM 层数据加工而成主要完成数据架构与整合建立一致性的维度构建可复用的面向分析和统计的明细事实表以及汇总公共粒度的指标。* DWDData Warehouse Detail 细节数据层是业务层与数据仓库的隔离层以业务过程作为建模驱动基于每个具体的业务过程特点构建细粒度的明细层事实表。可以结合企业的数据使用特点将明细事实表的某些重要维度属性字段做适当冗余也即宽表化处理。* DWSData WareHouse Service 服务数据层基于 DWD 的基础数据整合汇总成分析某一个主题域的服务数据以分析的主题为建模驱动基于上层的应用和产品的指标需求构建公共粒度的汇总指标事实表* 公共维度层DIM基于维度建模理念思想建立一致性维度* TMP 层临时层存放计算过程中临时产生的数据需要注意的是数据仓库层次划分不是固定不变的可以根据实际需求进行适当裁剪或者添加如果业务相对简单和独立可以将 DWD、DWS 进行合并。ADS层处理------ADSApplication Data Store 应用层数据。基于 DW 数据整合汇总成主题域的服务数据用于提供后续的业务查询等。### 数据明细层从数据明细层分析结果到 ClickHouse、Redis、Druid 等写入到 Redis 中。 class MySinkToRedis extends RichSinkFunction[(CityOrder, Long)] { private var jedis: Jedis ; override def open(parameters: Configuration): Unit { jedis new Jedis(“h121.wzk.icu”, 6379, 6000); jedis.select(0); // 选择 Redis 的第 0 个数据库 } override def invoke(value: (CityOrder, Long), context: SinkFunction.Context[]): Unit { if (!jedis.isConnected) { jedis.connect(); } val map new HashMapString, String; map.put(“totalMoney”, value._1.totalMoney.toString); map.put(“totalCount”, value._1.totalCount.toString); map.put(“time”, value._2.toString); // 打印信息用于调试 println(s”v a l u e . 1 . p r o v i n c e {value._1.province}value.1​.province{value._1.city} 数据条目: ${map.size()}, 金额: ${map.get(“totalMoney”)}, 数量: ${map.get(“totalCount”)}, 时间:KaTeX parse error: Expected }, got EOF at end of input: … jedis.hset(s{value._1.province}${value._1.city}, map); map.clear(); } catch { case e: Exception e.printStackTrace(); } } override def close(): Unit { if (jedis ! null) { jedis.close(); } } }