StreamingPro 中文文档

1. 五分钟快速上手和体验
2. Five Minute Quick Tutorial

应用模式和服务模式

1. 应用模式：写json配置文件，StreamingPro启动后执行该文件，可以作为批处理或者流式程序。
2. 服务模式：启动一个StreamingPro Server作为常驻程序,然后通过http接口发送MLSQL脚本进行交互。

我们强烈推荐使用第二种模式，第一种模式现在已经不太更新了，现在迅速迭代的是第二种模式，并且第二种模式可以构建AI平台。 为了避免编译的麻烦，你可以直接使用 release版本

对于确实需要使用json配置文件的，我们也提供了batch.mlsql脚本，可以让你使用mlsql语法，例如（v1.1.2开始具有这个功能）：

{"mlsql": {"desc":"测试","strategy":"spark","algorithm": [],"ref": [],"compositor": [
      {"name":"batch.mlsql","params": [
          {"sql": ["select 'a' as a as table1;","save overwrite table1 as parquet.`/tmp/kk`;"]
          }
        ]
      }
    ],"configParams": {
    }
  }
}

编译

• 编译文档
• 编译DSL文档

高级编程

• 自定义基于Python的算法模型

使用MLSQL做机器学习

• MLSQL使用SKLearn做文本分类实例
• MLSQL-分布式算法 based on spark MMLib
• MLSQL-深度学习 based on TensorFlow
• MLSQL-单机算法 based on SKLearn
• MLSQL日志回显

部署模型API服务

• 如何把模型部署成API服务

MLSQL常用功能

• MLSQL语法说明

• 各数据源读取和写入

• 如何使用CarbonData作为存储

• 常见特征处理模型说明

• 常见向量操作Vector操作函数

• 如何在MLSQL中运用分词抽词工具

• 网页抓取

• 批处理计算

• 流式计算

使用配置完成Spark编程

1. Spark 批处理
2. Spark Streaming
3. Structured Streaming
4. StreamingPro对机器学习的支持

概览：

1. 概述
2. 项目模块说明
3. 编译
4. 相关概念
5. StreamingPro的一些参数

周边工具

1. StreamingPro Manager
2. StreamingPro json文件编辑器支持

实验

1. flink支持

其他文档

• 英文文档
• 简书专栏
• 源码阅读-深入浅出StreamingPro
• 为什么开发MLSQL
• SQL服务

概述

StreamingPro 支持以Spark,Flink等作为底层分布式计算引擎，通过一套统一的配置文件完成批处理，流式计算，Rest服务的开发。 特点有：

1. 使用Json描述文件完成流式，批处理的开发，不用写代码。
2. 支持SQL Server,支持XSQL/MLSQL（重点），完成批处理，机器学习，即席查询等功能。
3. 标准化输入输出，支持UDF函数注册，支持自定义模块开发
4. 支持Web化管理Spark应用的启动，监控

如果更细节好处有：

1. 跨版本：StreamingPro可以让你不用任何变更就可以轻易的运行在spark 1.6/2.1/2.2上。
2. 新语法：提供了新的DSl查询语法/Json配置语法
3. 程序的管理工具：提供web界面启动/监控 Spark 程序
4. 功能增强：2.1之后Structured Streaming 不支持kafka 0.8/0.9 ,Structured，此外还有比如spark streaming 支持offset 保存等
5. 简化Spark SQL Server搭建成本：提供rest接口/thrift 接口，支持spark sql server 的负载均衡，自动将driver 注册到zookeeper上
6. 探索更多的吧

项目模块说明

相关概念

如果你使用StreamingPro，那么所有的工作都是在编辑一个Json配置文件。通常一个处理流程，会包含三个概念：

1. 多个输入
2. 多个连续/并行的数据处理
3. 多个输出

StreamingPro会通过'compositor'的概念来描述他们，你可以理解为一个处理单元。一个典型的输入compositor如下：

    {
        "name": "batch.sources",
        "params": [
          {
            "path": "file:///tmp/hdfsfile/abc.txt",
            "format": "json",
            "outputTable": "test"

          },
           {
              "path": "file:///tmp/parquet/",
              "format": "parquet",
              "outputTable": "test2"

            }
        ]
}

batch.sources就是一个compositor的名字。 这个compositor 把一个本地磁盘的文件映射成了一张表，并且告知系统，abc.txt里的内容 是json格式的。这样，我们在后续的compositor模块就可以使用这个 test表名了。通常，StreamingPro希望整个处理流程， 也就是不同的compositor都采用表来进行衔接。

StreamingPro不仅仅能做批处理，还能做流式，流式支持Spark Streaming,Structured Streaming。依然以输入compositor为例，假设 我们使用的是Structured Streaming,则可以如下配置。

    {
        "name": "ss.sources",
        "params": [
          {
            "format": "kafka9",
            "outputTable": "test",
            "kafka.bootstrap.servers": "127.0.0.1:9092",
            "topics": "test",
            "path": "-"
          },
          {
            "format": "com.databricks.spark.csv",
            "outputTable": "sample",
            "header": "true",
            "path": "/Users/allwefantasy/streamingpro/sample.csv"
          }
        ]
      }

第一个表示我们对接的数据源是kafka 0.9，我们把Kafka的数据映射成表test。 因为我们可能还需要一些元数据，比如ip和城市的映射关系， 所以我们还可以配置一些其他的非流式的数据源，我们这里配置了一个smaple.csv文件，并且命名为表sample。

如果你使用的是kafka >= 1.0,则 topics 参数需要换成'subscribe',并且使用时可能需要对内容做下转换，类似：

    select CAST(key AS STRING) as k, CAST(value AS STRING) as v from test

启动时，你需要把-streaming.platform 设置为 ss。

如果我们的输入输出都是Hive的话，可能就不需要batch.sources/batch.outputs 等组件了，通常一个batch.sql就够了。比如：

    "without-sources-job": {
    "desc": "-",
    "strategy": "spark",
    "algorithm": [],
    "ref": [],
    "compositor": [
      {
        "name": "batch.sql",
        "params": [
          {
            "sql": "select * from hiveTable",
            "outputTableName": "puarquetTable"
          }
        ]
      },
      {
        "name": "batch.outputs",
        "params": [
          {
            "format": "parquet",
            "inputTableName": "puarquetTable",
            "path": "/tmp/wow",
            "mode": "Overwrite"
          }
        ]
      }
    ],
    "configParams": {
    }
  }

在批处理里，batch.sources/batch.outputs 都是可有可无的，但是对于流式程序，stream.sources/stream.outputs/ss.sources/ss.outputs 则是必须的。

StreamingPro的一些参数

后面三个参数值得进一步说明：

假设我们定义了两个数据源，firstSource,secondSource,描述如下：

    {
        "name": "batch.sources",
        "params": [
          {
            "name":"firstSource",
            "path": "file:///tmp/sample_article.txt",
            "format": "com.databricks.spark.csv",
            "outputTable": "article",
            "header":true
          },
          {
              "name":"secondSource",
              "path": "file:///tmp/sample_article2.txt",
              "format": "com.databricks.spark.csv",
              "outputTable": "article2",
              "header":true
            }
        ]
      }

我们希望path不是固定的，而是启动时候决定的，这个时候，我们可以在启动脚本中使用-streaming.sql.source.[name].[参数] 来完成这个需求。 比如：

    -streaming.sql.source.firstSource.path  file:///tmp/wow.txt

这个时候，streamingpro启动的时候会动态将path 替换成你要的。包括outputTable等都是可以替换的。

有时候我们需要定时执行一个任务，而sql语句也是动态变化的，具体如下：

    {
        "name": "batch.sql",
        "params": [
          {
            "sql": "select * from test where hp_time=:today",
            "outputTableName": "finalOutputTable"
          }
        ]
      },

这个时候我们在启动streamingpro的时候，通过参数：

    -streaming.sql.params.today  "2017"

动态替换 sql语句里的:today