API接收地址

https://pipeline.qiniu.com

API返回内容

响应报文

• 如果请求成功,返回HTTP状态码200:

HTTP/1.1 200 OK

• 如果请求失败,返回包含如下内容的JSON字符串（已格式化,便于阅读）:

{
    "error":   "<errMsg    string>"
}

• 如果请求包含数据获取,则返回相应数据的JSON字符串；

创建消息队列(数据源)

请求语法

POST /v2/repos/<RepoName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "region": <Region>,
    "schema": [
      {
        "key": <Key>,
        "valtype": <ValueType>,
        "elemtype": <ElemType>,
        "required": <Required>,
        "schema": [
            ...
        ]
      },
      ...
    ],
    "options":{
      "withIP":<ipkeyname>
    }
}

请求内容

示例

curl -X POST https://pipeline.qiniu.com/v2/repos/Test_Repo \
-H 'Content-Type: application/json' \
-H 'Authorization: Pandora 2J1e7iG13J66GA8vWBzZdF-UR_d1MF-kacOdUUS4:NTi3wH_WlGxYOnXsvgUrO4XMD6Y='  \
-d '{
    "region": "nb", 
    "schema": [
        {
            "key": "userName",
            "valtype": "string",
            "required": true
        },
        {
            "key": "age",
            "valtype": "float",
            "required": true
        },
        {
            "key": "addresses",
            "valtype": "array",
            "elemtype": "long",
            "required": true
        },
        {
            "key": "profile",
            "valtype": "map",
            "required": true,
            "schema": [
              {
                  "key": "position",
                  "valtype": "string",
                  "required": true
              },
              {
                  "key": "salary",
                  "valtype": "float",
                  "required": true
              },
              {
                  "key": "education",
                  "valtype": "array",
                  "elemtype": "string",
                  "required": false
              }
            ]
        }
    ]
}'

查看所有消息队列

请求语法

GET /v2/repos
Authorization: Pandora <auth>

响应报文

Content-Type: application/json
{
    "repos": [
      {
        "name": <RepoName>,
        "region": <Region>,
        "derivedFrom": <TransformName>
      },
      ...
    ]
}

响应内容

根据名称查询消息队列

请求语法

GET /v2/repos/<RepoName>
Authorization: Pandora <auth>

响应报文

Content-Type: application/json
{
    "region": <Region>,
    "derivedFrom": <TransformName>,
    "schema": [
      {
        "key": <Key>,
        "valtype": <ValueType>,
        "elemtype": <ElemType>,
        "required": <Required>,
        "schema": [
            ...
        ]
      },
      ...
    ]
}

根据名称删除消息队列

请求语法

DELETE /v2/repos/<RepoName>
Authorization: Pandora <auth>

根据名称更新消息队列

请求语法

PUT /v2/repos/<RepoName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "schema": [
      {
        "key": <Key>,
        "valtype": <ValueType>,
        "elemtype": <ElemType>,
        "required": <Required>,
        "schema": [
            ...
        ]
      },
      ...
    ]
}

!> 注意： 更新字段信息时，如果需要保留已有的字段信息，也需要填写上去，这是一次全量更新。

创建流式计算任务

请求语法

POST /v2/repos/<RepoName>/transforms/<TransformName>/to/<DestinationRepoName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "plugin": {
    "name": <PluginName>,
    "output": [
      {
        "name": <FieldName1>,
        "type": <FieldType>
      },
      {
        "name": <FieldName2>
      },
      ......
    ],
  },
  "mode": <Mode>,
  "code": <Code>,
  "interval": <Interval>,
  "container": {
       "type": <ContainerType>,
       "count": <ContainerCount>
  },
  "whence": <TransformWhence>,
  "destrepo": [
      {
        "key": <Key>,
        "valtype": <ValueType>,
        "elemtype": <ElemType>,
        "required": <Required>,
        "schema": [
            ...
        ]
      },
      ...
  ]
}

请求内容

!> 注意:mode加code是基础的数据计算方式,自定义计算(plugin)是更为高级的数据计算方式,要注意mode/code和自定义计算两种计算方式可以共存,但不可以一种都不指定。当自定义计算和mode/code共存时,系统优先执行自定义计算,后执行mode/code。

示例

curl -X POST https://pipeline.qiniu.com/v2/repos/test_repo/transforms/transform_job/to/compute_repo \
     -H 'Content-Type: application/json' \
     -H 'Authorization: Pandora 2J1e7iG13J66GA8vWBzZdF-UR_d1MF-kacOdUUS4:NTi3wH_WlGxYOnXsvgUrO4XMD6Y=' \
     -d {
          "mode": "sql",
          "code": "select count(*) from test_repo",
          "interval": "1m",
          "container": {
               "type": "M16C4",
               "count": 5
              }
        }

上传jar包(自定义计算使用)

请求语法

POST /v2/plugins/<PluginName>
Content-Type: application/java-archive
Content-MD5: <ContentMD5>
Authorization: Pandora <auth>

请求内容

Plugin说明:

• Jar包的命名必须和包含代码方法的类名一致
• 上传的Plugin Jar包最大为100MB。
• Content-MD5头部是可选的。如果上传plugin的时候带上该头部服务器会校验上传数据的校验和,如果两者不一致服务器将拒绝上传。如果不带该头部,服务器不做任何校验和的检查。
• 是先计算plugin内容的MD5,再对MD5做一次base64编码转化为字符串。例如qiniu这个字符串的Content-MD5是gLL29S04bTCxYd2kCqsEIQ==而不是7b9d6b4d89f6825a196d4cc50fdbedc5
• PluginName必须与用户所编写的Parser类的全限定名保持一致,否则transform执行plugin会失败。 例如NginxLogParser位于com.qiniu包,PluginName须写为com.qiniu.NginxLogParser。

示例

curl -X POST https://pipeline.qiniu.com/v2/plugins/ComputeSumDataParser \
-H 'Content-Type: application/java-archive'  \
-H 'Content-MD5: 900150983cd24fb0d6963f7d28e17f72'  \
-H 'Authorization: Pandora 2J1e7iG13J66GA8vWBzZdF-UR_d1MF-kacOdUUS4:NTi3wH_WlGxYOnXsvgUrO4XMD6Y=' \
-T ./TestPlugin.jar \

查看所有流式计算任务

请求语法

GET /v2/repos/<RepoName>/transforms
Authorization: Pandora <auth>

查看指定流式计算任务的信息

请求语法

GET /v2/repos/<RepoName>/transforms/<TransformName>
Authorization: Pandora <auth>

响应报文

200 OK
Transform-Type: application/<TransformType>
{
    "name": "<TransformName1>",
    "to": "<DestRepo1>",
    "spec": {
      "plugin": {
        "name": <PluginName>,
        "output": [
          {
            "name": <FieldName1>,
            "type": <FieldType>
          },
          {
            "name": <FieldName2>
          },
          ...
        ],
      },
      "mode": <Mode>,
      "code": <Code>,
      "interval": <Interval>
    }
}

修改流式计算任务

请求语法

PUT /v2/repos/<RepoName>/transforms/<TransformName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "plugin": {
      "name": <PluginName>,
      "output": [
        {
          "name": <FieldName1>,
          "type": <FieldType>
        },
        {
          "name": <FieldName2>
        },
        ...
      ],
    },
    "code": <Code>,
    "interval": <Interval>,
    "container": {
      "type":  <ContainerType>,
      "count": <ContainerCount>
    }
}

注意:

1.更新计算任务,可以同时更新plugin、sql代码、运行时间间隔和配额,也可以只更新其中一种,但不能一种都不指定。

2.在更新sql和plugin的输出字段时,只能添加新的字段,不能删除和更改已经存在的字段。

删除流式计算任务

请求语法

DELETE /v2/repos/<RepoName>/transforms/<TransformName>
Authorization: Pandora <auth>

流式导出数据至HTTP地址

请求语法

POST /v2/repos/<RepoName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "type": <http>,
  "whence": <ExportWhence>,
  "spec": {
        "host": <Host>,      
        "uri": <RequestURI>,
        "format": <ExportFormat>  
    }
}

请求内容

!> 注意: 导出数据格式和推送数据相同。

示例

curl -X POST https://pipeline.qiniu.com/v2/repos/test_Repo/exports/export_job1 \
-H 'Content-Type: application/json' \
-H 'Authorization: Pandora 2J1e7iG13J66GA8vWBzZdF-UR_d1MF-kacOdUUS4:NTi3wH_WlGxYOnXsvgUrO4XMD6Y=' \
-d '{
  "type": "http",
  "spec": {
    "host": "www.qiniu.com",
    "uri": "/test/repos"
  }
}'

流式导出数据至时序数据库

请求语法

POST /v2/repos/<RepoName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "type": <tsdb>,
  "whence": <ExportWhence>,
  "spec": {
        "destRepoName": <DestRepoName>,
        "series": <SeriesName>,
        "omitInvalid": <OmitInvalid>,
        "omitEmpty": <OmitEmpty>,
        "tags": {
            "tag1": <#key1>,
            "tag2": <#key2>,
            ...
        },
        "fields": {
            "field1": <#key1>,
            "field2": <#key2>,
            ...
        },
        "timestamp": <#key1>,   
     }
}

请求内容

时序数据库中的timestamp字段的类型必须为 date；
消息队列中字段类型为:Long/String/Date 的字段都可以导出至时序数据库中的 timestamp 字段

示例

curl -X POST https://pipeline.qiniu.com/v2/repos/test_Repo/exports/export_job4 \
-H 'Content-Type: application/json' \
-H 'Authorization: Pandora 2J1e7iG13J66GA8vWBzZdF-UR_d1MF-kacOdUUS4:NTi3wH_WlGxYOnXsvgUrO4XMD6Y=' \
-d '{
  "type": "tsdb",
  "spec":{
          "destRepoName": "test_tsdb",
        "series": "req_io",
        "tags": {"type": "#type","src": "#src","zone": "#zone","time": "#time","bucket": "#bucket","domain": "#domain"
        },
        "fields": {"hits": "#hits","flow": "#flow"},            
    }
}'

流式导出数据至日志检索服务

请求语法

POST /v2/repos/<RepoName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "type": <logdb>,
  "whence": <ExportWhence>,
  "spec": {
        "destRepoName": <DestRepoName>,
        "omitInvalid": <OmitInvalid>,
        "omitEmpty": <OmitEmpty>,
        "doc": {
            "toRepoSchema1": <#fromRepoSchema1>,
            "toRepoSchema2": {
                "toRepoSchema3": <#fromRepoSchema3>,
            }
            ......
        }
}

请求内容

消息队列中,字段的类型与日志检索服务中的字段类型需要作出如下对应:

消息队列类型:string 对应 日志检索服务:string / date

消息队列类型:long 对应 日志检索服务:long / date

消息队列类型:float 对应 日志检索服务:float

消息队列类型:array[string] 对应 日志检索服务:string

消息队列类型:array[long] 对应 日志检索服务:long

消息队列类型:array[float] 对应 日志检索服务:float

消息队列类型:map 对应 日志检索服务:object

消息队列类型:date 对应 日志检索服务:date

消息队列类型:jsonstring 对应 日志检索服务:object

!> 注意: 对于消息队列的jsonstring类型，导出至日志检索服务的object类型时，会将内嵌字段名称中所有圆点(.)替换为下划线（由于日志检索服务的object类型的内嵌字段名称不支持圆点(.))。例如，消息队列中某一个字段f1的类型为jsonstring，取值为{"education.level": "university"}，最终导出至日志检索服务为{"education_level": "university"}，内嵌字段education.level中的圆点(,)被替换为下划线(_)。

示例

curl -X POST https://pipeline.qiniu.com/v2/repos/test_Repo/exports/export_job2 \
-H 'Content-Type: application/json' \
-H 'Authorization: Pandora 2J1e7iG13J66GA8vWBzZdF-UR_d1MF-kacOdUUS4:NTi3wH_WlGxYOnXsvgUrO4XMD6Y=' \
-d '{
  "type": "logdb",
  "spec": {
      "destRepoName": "logdb_testRepo", 
      "doc":{
            "user":"userName",
            "profile":{
                "age":"age"
            }
      }
    }
}'

流式导出数据至对象存储服务

请求语法

POST /v2/repos/<RepoName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "type": <kodo>,
    "whence": <ExportWhence>,
    "spec": {
         "bucket": <Bucket>,
         "keyPrefix": <KeyPrefix>, 
         "email": <Email>,  
         "accessKey": <AccessKey>,    
         "fields": {
             "key1": <#value1>,
             "key2": <#value2>,
             ...
         },
         "rotateStrategy": <RotateStrategy>,
         "rotateSize": <RotateSize>,
         "rotateInterval": <RotateInterval>,   
         "format": <Format>,
         "delimiter": <Delimiter>,
         "compress": <true|false>,
         "retention": <Retention>
    }
}

请求内容

!> 注1: compress 会压缩成gzip格式,但当用户指定format为parquet时,由于parquet已经是压缩好的列存格式,compress选项将不起作用。

!> 注2: keyPrefix字段表示导出文件名称的前缀,该字段可选,默认值为””(生成文件名会自动加上时间戳格式为yyyy-MM-dd-HH-mm-ss),如果使用了一个或者多个魔法变量时不会自动添加时间戳,支持魔法变量,采用$(var)的形式求值,目前可用的魔法变量var如下:

• year 上传时的年份
• mon 上传时的月份
• day 上传时的日期
• hour 上传时的小时
• min 上传时的分钟
• sec 上传时的秒钟

举例说明:

• 假如keyPrefix取值为kodo-parquet/date=$(year)-$(mon)-$(day)/hour=$(hour)/min=$(min)/$(sec),且生成某一文件时的北京标准时间为2017-01-12 15:30:00, 则keyPrefix将被解析为kodo-parquet/date=2017-01-12/hour=15/min=30/00,其中的魔法变量$(year)、$(mon)、$(day)、$(hour)、$(min)、$(sec)分别对应文件生成时间2017-01-12 15:30:00的年、月、日、时、分、秒。
• 假如keyPrefix使用默认值,且生成某一文件时的北京标准时间为2017-01-12 15:30:00, 则keyPrefix将被解析为2017-01-12-15-30-00。

示例

curl -X POST https://pipeline.qiniu.com/v2/repos/repox/exports/export1 \
-H 'Content-Type: application/json' \
-H 'Authorization: Pandora 2J1e7iG13J66GA8vWBzZdF-UR_d1MF-kacOdUUS4:NTi3wH_WlGxYOnXsvgUrO4XMD6Y='  \
-d '{
    "type": "kodo", 
    "spec": { 
        "bucket": "bucket2", 
        "keyPrefix": "key1/", 
        "email": "xiaoming@qiniu.com",
        "accessKey": "7H4FDc1M-FxXFZKwjbN_Up1OfY7DotXDjaM5jXzm",
        "fields": { 
            "f1": "#f1", 
            "f2": "#f2" 
        } 
        "format":"json",
        "compress": true
}'

流式导出数据至报表平台

请求语法

POST /v2/repos/<RepoName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "type": <report>,
  "whence": <ExportWhence>,
  "spec": {
        "dbName": <DBName>,
        "tableName": <TableName>,
        "columns": {
            "column1": <#key1>,
            "column2": <#key2>,
            ...
        }
     }
}

请求内容

更新流式导出任务

请求语法

PUT /v2/repos/<RepoName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "spec": <Spec>
}

!> 注意:仅支持对 spec 的修改（不允许修改type、whence）

查看所有流式导出任务

请求语法

GET /v2/repos/<RepoName>/exports
Authorization: Pandora <auth>

响应报文

Content-Type: application/json
{
    "name": <ExportName>,
    "type": <ExportSchema>,
    "spec": <Spec>,
    "whence": <ExportWhence>
}

根据名称查看流式导出任务

请求语法

GET /v2/repos/<RepoName>/exports/<ExportName>
Authorization: Pandora <auth>

根据名称删除流式导出任务

请求语法

DELETE /v2/repos/<RepoName>/exports/<ExportName>
Authorization: Pandora <auth>

错误代码及相关说明