场景

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

方案对比

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

方案	实时入库	SQL支持度
Spark+CarbonData	支持	Spark SQL语法丰富
Kylin	不支持	支持join
Flink+Druid	支持	0.15以前不支持SQL，不支持join

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

设计方案

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

 

场景举例

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

flink的处理

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

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

Druid处理

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

{
  "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": {
        "