一、DLA介绍

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

终于，阿里云现在也有了自己的数据湖分析产品（Data Lake Analytics，后续简称DLA）：https://www.aliyun.com/product/datalakeanalytics

产品文档：https://help.aliyun.com/product/70174.html

二、本案背景

因为一般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: 147
result_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: NULL
1 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__（https://www.aliyun.com/product/ons）服务，后续还会考虑支持kafka（https://www.aliyun.com/product/kafka）、mns（https://www.aliyun.com/product/mns）、http等接口或服务；

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

三、准备工作

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

进入https://www.aliyun.com/product/ons 开通服务，然后进入ons的控制台：https://ons.console.aliyun.com/

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

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

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

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

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

1. 给DLA云账号授权：

访问连接：https://ram.console.aliyun.com/#/role/authorize?request=%7B%22Requests%22:%20%7B%22request1%22:%20%7B%22RoleName%22:%20%22AliyunOpenAnalyticsAccessingMQRole%22,%20%22TemplateId%22:%20%22MQRole%22%7D%7D,%20%22ReturnUrl%22:%20%22https:%2F%2Fopenanalytics.console.aliyun.com%2F%22,%20%22Service%22:%20%22OpenAnalytics%22%7D

2. 在访问控制（https://ram.console.aliyun.com）中，查看自己的角色列表：

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: 147
result_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: 1834
1 row in set (0.02 sec)

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

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

d）消息消费：

消息是有3天保质期的，过期了消息就会被清理掉，请及时消费掉。具体的消息消费过程，请查看文档：https://help.aliyun.com/document_detail/29551.html，这里我不再赘述。