目前，实时计算越来越被广泛应用，比如 实时ETL、实时报表、实时大屏展示等一些监控预警和在线系统的场景。企业对计算速度和消息更新速度要求越来越高。开源框架中，Storm，Sparks，Flink等在企业生产中大量投入使用，但是开发相对复杂，需要对接各种框架api、sdk等，另外人力成本相对较高。那么有没有一种高效的实时计算平台，只要会写SQL并且可视化的操作就可以快速完成实时计算的业务开发呢
阿里云目前推出产品 阿里云流计算（公测中，预计18年3月份商业化）（StreamCompute）传送门：https://help.aliyun.com/document_detail/62437.html?spm=a2c4g.11186623.6.544.Ed7XzG
阿里云流计算全链路示意图
可以简单快速的实现仅用SQL就完成流计算的业务链路，下面我们就使用流计算给大家示范。
架构：
以阿里云流计算为核心，从数据流向上我们可以分为数据从哪里，到哪里去。
本次实验架构为：
Logstash+DataHub+阿里云流计算+RDS-mysql
其中，
Logstash ：开源框架，用于采集数据
DataHub：阿里云自主研发大型缓存队列（可以理解为类似 开源Kafka ）
https://help.aliyun.com/document_detail/47439.html?spm=a2c4g.11186623.6.539.hri7Gy
RDS-mysql：阿里云关系型数据库 mysql版
搭建流程：
第一步：创建DataHub 项目和Topic
具体参考
https://help.aliyun.com/document_detail/47448.html?spm=a2c4g.11186623.6.546.eJLHOm
如图是本次实验创建的：
第二步：搭建Logstash 参考
https://help.aliyun.com/document_detail/47451.html?spm=5176.product27797.6.588.iFTE4i
配置文件如下
input {
file {
path => "/Users/yang/test/stream.csv"
start_position => "beginning"
}
}
filter{
csv {
columns => ['name', 'age']
}
}
output {
datahub {
access_id => "LTAIu****Ouj87b"
access_key => "MfY8ONjK6******7OEdyXw4T"
endpoint => "https://dh-cn-hangzhou.aliyuncs.com/"
project_name => "M_shangdantest"
topic_name => "to_stream_topic"
#shard_id => "0"
#shard_keys => ["thread_id"]
dirty_data_continue => true
dirty_data_file => "/Users/yang/test/dirty.data"
dirty_data_file_max_size => 1000
}
}
第三步：创建流计算任务
1，可视化注册刚才创建的DataHub数据源，接入数据。如图
2，可视化注册数据输出的数据库RDS-mysql
3，编写从数据来源（DataHub）取数据经过计算写入数据目的源（RDS-mysql）的业务SQL。
如图
代码附上，具体业务逻辑可以自己通过SQL实现：
CREATE TABLE to_stream_topic (
`name` VARCHAR,
age BIGINT
) WITH (
type = 'datahub',
endPoint = 'http://dh-cn-hangzhou.aliyun-inc.com/',
roleArn='acs:ram::xxxxx:role/aliyunstreamdefaultrole',
project = 'shangdantest',
topic = 'to_stream_topic'
);
INSERT INTO resoult_stream
SELECT name,age from to_stream_topic;
CREATE TABLE resoult_stream (
`name` VARCHAR,
age BIGINT
) WITH (
type= 'rds',
url = 'jdbc:mysql://rm-xxxxxxx.mysql.rds.aliyuncs.com:3306/lptest',
userName = 'xxxxx',
password = 'xxxx',
tableName = 'resoult_stream'
);
更多操作其他数据源参考：
https://help.aliyun.com/document_detail/62509.html?spm=a2c4g.11186623.6.633.DPannE
以上操作完成后可以
1，启动logstash
2，测试流计算，调试非常方便，对线上业务没有影响，自动从DataHub中抽取数据进行测试
3，启动流计算作业，如图
4，观察数据是否成功写入RDS-Mysql
到此 流计算一个实时链路搭建完毕，有没有发现很酷炫，只需要写写SQL，加一些数据源的配置即可。整个过程一小时就可以完成，相对开源省去了繁琐的各种环境搭建，代码编写，监控等等。
有对大数据技术感兴趣的，可以加笔者的微信 wx4085116.目前笔者已经从阿里离职，博客不代表阿里立场。笔者开了一个大数据培训班。有兴趣的加我。
