Flink+Druid构建实时OLAP的探索

柔情只为你懂 2021-11-02 16:28 455阅读 0赞

#### 场景 ####

k12在线教育公司的业务场景中，有一些业务场景需要实时统计和分析，如分析在线上课老师数量、学生数量，实时销售额，课堂崩溃率等，需要实时反应上课的质量问题，以便于对整个公司的业务情况有大致的了解。

#### 方案对比 ####

对比了很多解决方案，如下几种，列出来供参考。

<table> 
 <thead> 
  <tr> 
   <th>方案</th> 
   <th>实时入库</th> 
   <th>SQL支持度</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Spark+CarbonData</td> 
   <td>支持</td> 
   <td>Spark SQL语法丰富</td> 
  </tr> 
  <tr> 
   <td>Kylin</td> 
   <td>不支持</td> 
   <td>支持join</td> 
  </tr> 
  <tr> 
   <td>Flink+Druid</td> 
   <td>支持</td> 
   <td>0.15以前不支持SQL，不支持join</td> 
  </tr> 
 </tbody> 
</table>

1.  [上一篇文章][Link 1]所示，使用Spark+CarbonData也是一种解决方案，但是他的缺点也是比较明显，如不能和Flink进行结合，因为我们整个的大数据规划的大致方向是，Spark用来作为离线计算，Flink作为实时计算，并且这两个大方向短时间内不会改变；
2.  Kylin一直是老牌OLAP引擎，但是有个缺点无法满足我们的需求，就是在技术选型的那个时间点kylin还不支持实时入库（后续2.0版本支持实时入库），所以就选择了放弃；
3.  使用Flink+Druid方式实现，这个时间选择这个方案，简直是顺应潮流呀，Flink现在如日中天，各大厂都在使用，Druid是OLAP的新贵，关于它的文章也有很多，我也不赘述太多。有兴趣的可以看下[这篇文章][Link 2]，我的博客其它文章也有最新版本的安装教程，实操方案哦。

#### 设计方案 ####

实时处理采用Flink SQL，实时入库Druid方式采用 [druid-kafka-indexing-service][]，另一种方式入库方式，[Tranquility][]，这种方式测试下来问题多多，放弃了。数据流向如下图。

![739637-20190809180703076-379504187.png][]

#### 场景举例 ####

实时计算课堂连接掉线率。此事件包含两个埋点上报，进入教室和掉线分别上报数据。druid设计的字段

##### flink的处理 #####

将上报的数据进行解析，上报使用的是json格式，需要解析出所需要的字段然后发送到kafka。字段包含如下

sysTime，DateTime格式 pt,格式yyyy-MM-dd eventId,事件类型(enterRoom|disconnect) lessonId,课程ID

##### Druid处理 #####

启动Druid Supervisor，消费Kafka里的数据，使用预聚合，配置如下

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

{
      "type": "kafka",
      "dataSchema": {
        "dataSource": "sac_core_analyze_v1",
        "parser": {
          "parseSpec": {
            "dimensionsSpec": {
              "spatialDimensions": [],
              "dimensions": [
                "eventId",
                "pt"
              ]
            },
            "format": "json",
            "timestampSpec": {
              "column": "sysTime",
              "format": "auto"
            }
          },
          "type": "string"
        },
        "metricsSpec": [
          {
                "filter": {
                    "type": "selector",
                    "dimension": "msg_type",
                    "value": "disconnect"
                },
                "aggregator": {
                    "name": "lesson_offline_molecule_id",
                    "type": "cardinality",
                    "fields": ["lesson_id"]
                },
                "type": "filtered"
            }, {
                "filter": {
                    "type": "selector",
                    "dimension": "msg_type",
                    "value": "enterRoom"
                },
                "aggregator": {
                    "name": "lesson_offline_denominator_id",
                    "type": "cardinality",
                    "fields": ["lesson_id"]
                },
                "type": "filtered"
            }
        ],
        "granularitySpec": {
          "type": "uniform",
          "segmentGranularity": "DAY",
          "queryGranularity": {
            "type": "none"
          },
          "rollup": true,
          "intervals": null
        },
        "transformSpec": {
          "filter": null,
          "transforms": []
        }
      },
      "tuningConfig": {
        "type": "kafka",
        "maxRowsInMemory": 1000000,
        "maxBytesInMemory": 0,
        "maxRowsPerSegment": 5000000,
        "maxTotalRows": null,
        "intermediatePersistPeriod": "PT10M",
        "basePersistDirectory": "/tmp/1564535441619-2",
        "maxPendingPersists": 0,
        "indexSpec": {
          "bitmap": {
            "type": "concise"
          },
          "dimensionCompression": "lz4",
          "metricCompression": "lz4",
          "longEncoding": "longs"
        },
        "buildV9Directly": true,
        "reportParseExceptions": false,
        "handoffConditionTimeout": 0,
        "resetOffsetAutomatically": false,
        "segmentWriteOutMediumFactory": null,
        "workerThreads": null,
        "chatThreads": null,
        "chatRetries": 8,
        "httpTimeout": "PT10S",
        "shutdownTimeout": "PT80S",
        "offsetFetchPeriod": "PT30S",
        "intermediateHandoffPeriod": "P2147483647D",
        "logParseExceptions": false,
        "maxParseExceptions": 2147483647,
        "maxSavedParseExceptions": 0,
        "skipSequenceNumberAvailabilityCheck": false
      },
      "ioConfig": {
        "topic": "sac_druid_analyze_v2",
        "replicas": 2,
        "taskCount": 1,
        "taskDuration": "PT600S",
        "consumerProperties": {
          "bootstrap.servers": "bd-prod-kafka01:9092,bd-prod-kafka02:9092,bd-prod-kafka03:9092"
        },
        "pollTimeout": 100,
        "startDelay": "PT5S",
        "period": "PT30S",
        "useEarliestOffset": false,
        "completionTimeout": "PT1200S",
        "lateMessageRejectionPeriod": null,
        "earlyMessageRejectionPeriod": null,
        "stream": "sac_druid_analyze_v2",
        "useEarliestSequenceNumber": false
      },
      "context": null,
      "suspended": false
    }

最重要的配置是metricsSpec，他主要定义了预聚合的字段和条件。

##### 数据查询 #####

数据格式如下

<table> 
 <thead> 
  <tr> 
   <th>pt</th> 
   <th>eventId</th> 
   <th>lesson_offline_molecule_id</th> 
   <th>lesson_offline_denominator_id</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>2019-08-09</td> 
   <td>enterRoom</td> 
   <td>"AQAAAAAAAA=="</td> 
   <td>"AQAAAAAAAA=="</td> 
  </tr> 
  <tr> 
   <td>2019-08-09</td> 
   <td>disconnect</td> 
   <td>"AQAAAAAAAA=="</td> 
   <td>"AQAAAAAAAA=="</td> 
  </tr> 
 </tbody> 
</table>

结果可以按照这样的SQL出

SELECT pt,CAST(APPROX_COUNT_DISTINCT(lesson_offline_molecule_id) AS DOUBLE)/CAST(APPROX_COUNT_DISTINCT(lesson_offline_denominator_id) AS DOUBLE) from sac_core_analyze_v1 group by pt

可以使用Druid的接口查询结果，肥肠的方便~

转载于:https://www.cnblogs.com/ChouYarn/p/11328900.html

[Link 1]: https://www.cnblogs.com/ChouYarn/p/11316517.html
[Link 2]: https://www.cnblogs.com/xd502djj/p/6408979.html
[druid-kafka-indexing-service]: https://druid.apache.org/docs/latest/development/extensions-core/kafka-ingestion.html
[Tranquility]: https://druid.apache.org/docs/latest/ingestion/stream-push.html
[739637-20190809180703076-379504187.png]: /images/20211101/daf9981568a546c280a6b89e04821869.png
[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20211101/6e6b0542cd34488ca571a9095b74857e.png