一、DLA介绍

数据湖（Data Lake）是时下热门的概念，更多阅读可以参考：。基于数据湖，可以不用做任何ETL、数据搬迁等过程，实现跨各种异构数据源进行大数据关联分析，从而极大的节省成本和提升用户体验。

终于，阿里云现在也有了自己的数据湖分析产品（Data Lake Analytics，后续简称DLA）：

产品文档：

二、本案背景

因为一般OLAP的场景，SQL通常比较复杂也比较耗时，如果程序一直阻塞等待的话，会影响业务，因此DLA支持异步执行任务。在SQL上加一个Hint即可，相关的查询结果会默认写入到您的OSS目录中（如果OSS未开通则无法使用异步查询，目录是DLA默认规划好的），并且会返回这个Query对应的ID回执给您，这个回执在后续有用：

mysql> /*+ run-async=true */select * from test_table1;+--------------------------------+| ASYNC_TASK_ID                  |+--------------------------------+| q201811021109sh8d1a0b750000182 |+--------------------------------+1 row in set (0.04 sec)

如果您想要查看您的异步SQL任务执行状况，可以执行另一个SQL：

mysql> show query_task where id = 'q201811021109sh8d1a0b750000182'\G*************************** 1. row ***************************                  id: q201811021109sh8d1a0b750000182        mpp_query_id: 20181102_030939_149_svnhw              status: SUCCESS           task_name: SELECT        table_schema: sh_tpch             command: /*+ run-async=true */select * from test_table1          creator_id: ${您的dla账号}         create_time: 2018-11-02 11:09:40.0         update_time: 2018-11-02 11:09:40.0       connection_id: 693929276088405             message:            row_count: 2         elapse_time: 692  scanned_data_bytes: 147result_file_oss_file: oss://aliyun-oa-query-results-${您的uid}-oss-cn-shanghai/DLA_Result/2018/11/02/q201811021109sh8d1a0b750000182/result.csv    cancellable_task: 0          mq_product: NULL            mq_topic: NULL      mq_producer_id: NULL            mq_model: NULL           mq_status: NULL        mq_error_msg: NULL       mq_message_id: NULL       mq_total_time: NULL1 row in set (0.02 sec)

其中status: SUCCESS（RUNNING表示运行中，FAILURE表示已失败，CANCELLED表示用户取消，等等）表示这个Query执行成功，result_file_oss_file: oss://aliyun-oa-query-results-.......就是存储到您的OSSQuery的结果集文件在您的OSS上，elapse_time: 692表示整体Query花费的时间，scanned_data_bytes: 147表示Query扫描的数据量（收费依据），等等信息；

但关键问题是，用户程序需要一直轮询执行show query_task，才能知道Query的状态是不是从RUNNING转移到SUCCESS状态，这样对用户的程序影响很大，因此需要有通知机制来及时告知用户程序。

所以，DLA最近支持了消息回执的机制，帮助用户快速、异步感知任务执行状态。目前DLA已经支持将消息写入到阿里云上的__ONS__（）服务，后续还会考虑支持kafka（）、mns（）、http等接口或服务；

下面，就以我们的测试账号，来介绍下如何在DLA中使用ONS来做消息回执通知。

三、准备工作

a）开通ONS服务，进入控制台

进入 开通服务，然后进入ons的控制台：

b）选择与DLA相同的Region，创建相关的元信息

1. 在这里，我们选择华东2集群，作为我们的测试集群：

2. 创建Topic，专门接收DLA的消息：

3. 为DLA创建一个生产者：

c）为DLA开通角色授权，允许DLA给您的topic发送消息

1. 给DLA云账号授权：

访问连接：

2. 在访问控制（）中，查看自己的角色列表：

3. 确认一下，确实是只有PUB权限：

四、开始使用：

a）目前消息回执，也是通过在SQL上增加hint的方式来通知：

mysql> /*+ run_async=true, mq-notify-by=ons, mq-topic=dla_shanghai_topic, mq-producer-id=PID_dla_shanghai */ select * from test_table1;+--------------------------------+| ASYNC_TASK_ID                  |+--------------------------------+| q201811021152sh8d1a0b750000191 |+--------------------------------+1 row in set (0.03 sec)

其中mq-notify-by=ons, mq-topic=dla_shanghai_topic, mq-producer-id=PID_dla_shanghai是上面已经配置过的。

b）执行show query_task看下效果：

mysql> show query_task where id = 'q201811021152sh8d1a0b750000191'\G*************************** 1. row ***************************                  id: q201811021152sh8d1a0b750000191        mpp_query_id: 20181102_035233_150_svnhw              status: SUCCESS           task_name: SELECT        table_schema: sh_tpch             command: /*+ run_async=true, mq-notify-by=ons, mq-topic=dla_shanghai_topic, mq-producer-id=PID_dla_shanghai */ select * from test_table1          creator_id: ${您的dla账号}         create_time: 2018-11-02 11:52:34.0         update_time: 2018-11-02 11:52:35.0       connection_id: 693929276092181             message:            row_count: 2         elapse_time: 814  scanned_data_bytes: 147result_file_oss_file: oss://aliyun-oa-query-results-${您的uid}-oss-cn-shanghai/DLA_Result/2018/11/02/q201811021152sh8d1a0b750000191/result.csv    cancellable_task: 0          mq_product: ons            mq_topic: dla_shanghai_topic      mq_producer_id: PID_dla_shanghai            mq_model: NULL           mq_status: SUCCESS        mq_error_msg: NULL       mq_message_id: AC1314094B444E0E2F2A07B2C1E40000       mq_total_time: 18341 row in set (0.02 sec)

上面的mq_xxx列，都是关于消息通知的信息。其中mq_status: SUCCESS表示消息发送成功，而mq_message_id: AC1314094B444E0E2F2A07B2C1E40000则是这个消息回执的消息id。

c）进入消息查询，查看消息情况：

d）消息消费：

消息是有3天保质期的，过期了消息就会被清理掉，请及时消费掉。具体的消息消费过程，请查看文档：，这里我不再赘述。