目前，实时计算越来越被广泛应用，比如实时ETL、实时报表、实时大屏展示等一些监控预警和在线系统的场景。企业对计算速度和消息更新速度要求越来越高。开源框架中，Storm，Sparks，Flink等在企业生产中大量投入使用，但是开发相对复杂，需要对接各种框架api、sdk等，另外人力成本相对较高。那么有没有一种高效的实时计算平台，只要会写SQL并且可视化的操作就可以快速完成实时计算的业务开发。

阿里云目前推出产品阿里云流计算(公测中，预计18年3月份商业化)(StreamCompute)阿里云流计算全链路示意图

可以简单快速的实现仅用SQL就完成流计算的业务链路，下面我们就使用流计算给大家示范。

架构：

以阿里云流计算为核心，从数据流向上我们可以分为数据从哪里，到哪里去。

本次实验架构为：

Logstash+DataHub+阿里云流计算+RDS-mysql

其中，

Logstash ：开源框架，用于采集数据

DataHub：阿里云自主研发大型缓存队列(可以理解为类似开源Kafka )

RDS-mysql：阿里云关系型数据库 mysql版

搭建流程：

第一步：创建DataHub 项目和Topic

具体参考

如图是本次实验创建的：

第二步：搭建Logstash

具体参考

配置文件如下

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'

);

以上操作完成后可以

1，启动logstash

2，测试流计算，调试非常方便，对线上业务没有影响，自动从DataHub中抽取数据进行测试

3，启动流计算作业，如图

4，观察数据是否成功写入RDS-Mysql

到此 流计算一个实时链路搭建完毕，有没有发现很酷炫，只需要写写SQL，加一些数据源的配置即可。整个过程一小时就可以完成，相对开源省去了繁琐的各种环境搭建，代码编写，监控等等。

了解更多请微博关注阿里云客户满意中心