调用方式

调用方式

API 接收地址

https://pipeline.qiniu.com

API 返回内容

响应报文

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

HTTP/1.1 200 OK

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

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

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

数据源相关接口

创建批量计算数据源

请求语法

 POST /v2/datasources/<DataSourceName>
 Content-Type: application/json
 Authorization: Pandora <auth>
 {
     "region": <Region>,
     "type": <Type>,
     "spec": <Spec>,
     "noVerifySchema": <NoVerifySchema>,
     "workflow": <WorkflowName>,
     "schema": [
       {
         "key": <Key>,
         "valtype": <ValueType>,
         "required": <Required>
       },
       ...
     ]
 }

参数说明

当 type 为hdfs的时候 spec 定义如下:

当type为kodo的时候 spec 定义如下:

当type为fusion的时候 spec 定义如下:

!> 注意：region参数的选择以降低传输数据的成本为原则，请尽量选择离自己数据源较近的区域。

更新批量计算数据源

请求语法

 PUT /v2/datasources/<DataSourceName>
 Content-Type: application/json
 Authorization: Pandora <auth>
 {
     "spec": <Spec>,
     "schema": [
       {
         "key": <Key>,
         "valtype": <ValueType>,
         "required": <Required>
       },
       ...
     ]
 }

!> 注意: 更新批量计算数据源字段信息的时候，不允许减少字段，也不允许更改字段的类型。

根据名称查看批量计算数据源

请求语法

GET /v2/datasources/<DataSourceName>
Authorization: Pandora <auth>

响应报文

 {
     "region": <Region>,
     "type": <Type>,
     "spec": <Spec>,
     "schema": [
       {
         "key": <Key>,
         "valtype": <ValueType>,
         "required": <Required>
       },
       ...
     ]
 }

列举批量计算数据源

请求语法

GET /v2/datasources
Authorization: Pandora <auth>

响应报文

 {
   "datasources": [
     {
       "name": <DataSourceName>,
       "region": <Region>,
       "type": <Type>,
       "spec": <Spec>,
       "schema": [
         {
           "key": <Key>,
           "valtype": <ValueType>,
           "required": <Required>
         },
         ...
       ]
     },
     ...
   ]
 }

根据名称删除批量计算数据源

请求语法

DELETE /v2/datasources/<DataSourceName>
Authorization: Pandora <auth>

批量计算任务相关接口

创建批量计算任务

请求语法

POST /v2/jobs/<JobName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "srcs":[
        {
            "name":<DataSourceName|JobName>,
            "type":<DataSource|Job>,
            "tableName": <TableName>
        },
        ...
    ],
   "computation": {
       "code": <Code>,
       "type": <SQL>
   }, 
   "container": {
       "type": <ContainerType>,
       "count": <ContainerCount>
   },  
   "scheduler":{
       "type": <crontab|loop|manual|depend>,
       "spec": {
           "crontab": <0 0 0/1 * *>,
           "loop": <1h|3m|....>
       } 
   },
   "params":[
       {
           "name":<ParamName>,
           "default":<ParamValue>
       },
       ...
   ]
}

参数说明

scheduler.type 为 depend 的时候，container 依赖上游的配置，该配置不填

scheduler.type 为 manual 和 depend 的时候 spec 可以不填

scheduler.type 为 loop时，不填该 spec.loop 或者该字段为 0，则默认持续运行该任务

scheduler.type 为 depend 的时候，params 依赖上游的配置，该配置不填

!> 注1：scheduler.type 如果是 depend 模式，代表这个离线任务依赖某个上游的离线任务。首先 srcs 内有且仅有一个离线任务数据源。同时该任务不能指定调度的模式、魔法变量和容器规格。这些全部使用上游依赖的离线任务。

!> 注2：srcs.tableName 在此任务的 srcs 中以及依赖的所有上游任务的 srcs 中不能重复。

文件前缀+文件过滤规则举例说明：

文件前缀：audit-log/xxx-log-

文件过滤规则： $(year)-$(mon)-$(day)

最终的文件地址为： audit-log/xxx-log-2017-06-01

更新批量计算任务

请求语法

PUT /v2/jobs/<JobName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "srcs":[
        {
            "name":<DataSourceName|JobName>,
            "fileFilter":<fileFilter>,
            "type":<DataSource|Job>,
            "tableName": <TableName>
        },
        ...
    ],
   "computation": {
       "code": <Code>,
       "type": <SQL>
   }, 
   "container": {
       "type": <ContainerType>,
       "count": <ContainerCount>
   },
   "scheduler":{
       "type": <crontab|loop|manual|depend>,
       "spec": {
           "crontab": <0 0 0/1 * *>,
           "loop": <1h|3m|....>
       }
   },
   "params":[
       {
           "name":<ParamName>,
           "default":<ParamValue>
       },
       ...
   ]
}

!> 注意：更新时候 srcs, code, container, scheduler, params 校验逻辑和创建的时候相同。下游计算任务不指定container, scheduler, params。更新逻辑为全量更新。需要提前获取所有信息。

列举批量计算任务信息

请求语法

GET /v2/jobs?srcDatasource=[DataSourceName]&srcJob=[JobName]
Authorization: Pandora <auth>

响应报文

{
    "jobs":[
        {
            "name": <JobName>,
            "srcs":[
                {
                    "name":<DataSourceName|JobName>,
                    "fileFilter":<fileFilter>,
                    "type":<DataSource|Job>,
                    "tableName": <TableName>
                },
                ...
            ],
           "scheduler":{
               "type": <crontab|loop|manual|depend>,
               "spec": {
                   "crontab": <0 0 0/1 * *>,
                   "loop": <1h|3m|....>
               }
           },
           "computation": {
               "code": <Code>,
               "type": <SQL>
           }, 
           "container": {
               "type": <ContainerType>, 
               "count": <ContainerCount>
           },
           "params":[
               {
                   "name":<ParamName>,
                   "default":<ParamValue>
               },
               ...
           ]，
           "schema": [
               {
                   "key": <Key>,
                   "valtype": <ValueType>
               },
               ...
           ]
        },
        ...
    ]
}

参数说明

!> 注意: srcDataSource 和 srcJob 不能同时存在。当两个参数都不指定的时候列举出所有的离线计算任务。

获取单个批量计算任务信息

请求语法

GET /v2/jobs/<JobName>
Content-Type: application/json
Authorization: Pandora <auth>

响应报文

{
    "srcs":[
        {
            "name":<DataSourceName|JobName>,
            "fileFilter":<fileFilter>,
            "type":<DataSource|Job>,
            "tableName": <TableName>
        },
        ...
    ],
   "scheduler":{
       "type": <crontab|loop|manual|depend>,
       "spec": {
           "crontab": <0 0 0/1 * *>,
           "loop": <1h|3m|....> 
          } 
    },
   "computation": {
       "code": <SqlCode>,
       "type": <SQL>
   }, 
   "container": {
       "type": <ContainerType>, 
       "count": <ContainerCount>
   },
   "params":[
           {
               "name":<ParamName>,
               "default":<ParamValue>
           },
           ...
   ]，
   "schema": [
      {
        "key": <Key>,
        "valtype": <ValueType>
      },
      ...
    ]
}

启动批量计算任务

请求语法

POST /v2/jobs/<JobName>/actions/start
Authorization: Pandora <auth>
{
    "params":[
           {
               "name":<ParamName>,
               "value":<ParamValue>
           },
           ...
   ]
}

参数说明

停止批量计算任务

请求语法

POST /v2/jobs/<JobName>/actions/stop
Authorization: Pandora <auth>

删除批量计算任务信息

请求语法

DELETE /v2/jobs/<JobName>
Authorization: Pandora <auth>

获取数据源schema

请求语法

POST /v2/schemas
Content-Type: application/json
Authorization: Pandora <auth>
{
    "type": <Kodo|HDFS|Fusion>, 
    "spec": <Spec>
}

参数说明

Type 为 Kodo 时，Spec结构：

Type 为 HDFS 时，Spec 结构：

Type为 Fusion 时，Spec 为空结构体.

响应内容

json or parquet return :
{
    "schema": [
        {
            "key": <Key>,
            "valtype": <ValueType>
        },
        ...
    ]
}
text return :
{
    "schema": [
        {
            "key": <text>,
            "valtype": <String>
        },
        ...
    ]
}

批量计算任务导出接口

导出数据至云存储

请求语法

POST /v2/jobs/<JobName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "type": <kodo>,
    "spec": {
         "bucket": <Bucket>,
         "keyPrefix": <Prefix|Path>,
         "format": <ExportFormat>,
         "delimiter": <Delimiter>,
         "containsHeader":<True|False>
         "compression": <compression>,
         "retention": <Retention>,
         "partitionBy": <PartitionBy>，
         "fileCount": <FileCount>,
         "saveMode": <SaveMode>
    }
}

参数说明

!> 注1: 当用户指定format为json、csv或text时, compression仅支持none(不压缩)、bzip2, gzip, lz4, snappy和deflate; 当用户指定format为orc时, compression仅支持none(不压缩)、snappy, zlib和lzo; 当用户指定format为parquet时, compression仅支持none(不压缩)、snappy, gzip和lzo。

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

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

导出数据至 HDFS

请求语法

POST /v2/jobs/<JobName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
    "type": <hdfs>,
    "spec": {
         "path": <path_to_write>,
         "format": <ExportFormat>,
         "delimiter": <Delimiter>,
         "compression": <compression>,
         "partitionBy": <PartitionBy>，
         "fileCount": <FileCount>,
         "saveMode": <SaveMode>
    }
}

参数说明

!> 注1: 当用户指定format为json、csv或text时, compression仅支持none(不压缩)、bzip2, gzip, lz4, snappy和deflate; 当用户指定format为orc时, compression仅支持none(不压缩)、snappy, zlib和lzo; 当用户指定format为parquet时, compression仅支持none(不压缩)、snappy, gzip和lzo。

!> 注2: path字段支持魔法变量,采用$(var)的形式求值,目前可用的魔法变量var如下:

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

导出数据至 HTTP 地址

请求语法

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

参数说明

!> 注1: 当导出格式为text时，导出的服务端需要支持Content-Type(MimeType)为text/plain的请求; 当导出格式为json时，导出服务端需要支持Content-Type(MimeType)为application/json的请求。 !> 注2: text的导出数据格式为：Key1=Value1\tKey2=Value2\tKey3=Value3; json的导出数据格式为：{"Key1": Value1, "Key2": Value2, "Key3": Value3}。

导出数据至日志检索服务

请求语法

POST /v2/jobs/<JobName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "type": <logdb>,
  "spec": {
        "destRepoName": <DestRepoName>,
        "omitInvalid": <OmitInvalid>,              
        "doc": {
            "LogdbRepoField1": <#JobField1>,
            "LogdbRepoField2": <#JobField2>,
            "LogdbRepoField3": <#JobField3>,         
            ......
        }
}

参数说明

导出数据至时序数据库服务

请求语法

POST /v2/jobs/<JobName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "type": <tsdb>,
  "spec": {
        "destRepoName": <DestRepoName>,
        "omitInvalid": <OmitInvalid>, 
        "series": <Series>,             
        "tags": {
            "tag1": <#JobField1>,
            "tag2": <#JobField2>,
            ...
        },
        "fields": {
            "field1": <#JobField3>,
            "field2": <#JobField4>,
            ...
        },
        "timestamp": <#JobField5> 
}

参数说明

导出数据至报表平台

请求语法

POST /v2/jobs/<JobName>/exports/<ExportName>
Content-Type: application/json
Authorization: Pandora <auth>
{
  "type": <report>,
  "spec": {
        "dbName": <DBName>,
        "tableName": <TableName>,
        "columns": {
            "column1": <#JobField1>,
            "column2": <#JobField2>,
            "column3": <#JobField3>,
            ...
        },
        "saveMode": <SaveMode>
}

参数说明

更新导出任务

请求语法

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

按照名称查看导出任务

请求语法

GET /v2/jobs/<JobName>/exports/<ExportName>
Authorization: Pandora <auth>

响应报文

{
    "type": <Type>,
    "spec": <Spec>
}

列举导出任务

请求语法

GET /v2/jobs/<JobName>/exports
Authorization: Pandora <auth>

响应报文

Content-Type: application/json
{
  "exports": [
    {
      "name": <ExportName>,
      "type": <Type>,
      "spec": <Spec>
    },
    ...
  ]
}

按照名称删除导出任务

请求语法

DELETE /v2/jobs/<JobName>/exports/<ExportName>
Authorization: Pandora <auth>

查看历史批量计算任务

请求语法

GET /v2/jobs/<jobName>/history?page=1&size=20&sortBy=endTime&status=xx&runId=-1
Content-Type: application/json
Authorization: Pandora <auth>

参数说明

响应报文

{
"total": <TotalCnt>, # 总共运行批次
"history":[
    {
        "runId" : <RunId>, # 运行批次
        "batchTime": "<batchTime>", #批次时间，每个批次第一次被调度的时间，是不可变的。
        "startTime": <StartTime>,
        "endTime": <EndTime>,
        "duration": <Duration>,
        "status": <Ready、Successful、Failed、Running、Canceled>,
        "message": <Message>
    },
  ]
}

参数说明

停止批次任务

请求语法

POST /v2/batch/actions/stop
Content-Type: application/json
Authorization: Pandora <auth>
{
    "jobName": "<jobName>",
    "runId": "<runId>"
}

参数说明

响应报文

{
  "preStatus": "Running",
  "postStatus": "Canceled"
}

参数说明

状态说明

重跑批次任务

请求语法

POST /v2/batch/actions/rerun
Content-Type: application/json
Authorization: Pandora <auth>
{
    "jobName": "<jobName>",
    "runId": "<runId>"
}

参数说明

响应报文

{
  "preStatus": "Failed",
  "postStatus": "Running",
  "rerunCount": 2
}

参数说明

状态说明

魔法变量接口

创建魔法变量

请求语法

 POST /v2/variables/<VariableName>
 Content-Type: application/json
 Authorization: Pandora <auth>
 {
     "type": <Type>,
     "value": <Value>,
     "format": <Format>,
 }

参数说明

!> 备注：系统默认可用的魔法变量如下:

• year 表示type为time；value为当前年份；format为 yyyy, 如2017
• mon 表示type为time；value为当前月份；format为 MM, 如01
• day 表示type为time；value为当前日期；format为 dd, 如01
• hour 表示type为time；value为当前小时；format为 HH, 如00
• min 表示type为time；value为当前分钟；format为 mm, 如00
• sec 表示type为time；value为当前秒数；format为 ss, 如00
• date 表示type为time；value为当前秒数；format为 yyyy-MM-dd, 如2017-01-01
• now 表示type为time；value为当前时间；format为 yyyy-MM-dd HH:mm:ss, 如2017-01-01 00:00:00

更新魔法变量

请求语法

 PUT /v2/variables/<VariableName>
 Content-Type: application/json
 Authorization: Pandora <auth>
 {
     "type": <Type>,
     "value": <Value>,
     "format": <Format>
 }

根据名称查看魔法变量

请求语法

GET /v2/variables/<VariableName>
Authorization: Pandora <auth>

响应报文

 {
     "name": <Name>,
     "type": <Type>,
     "value": <Value>,
     "format": <Format>
 }

列举魔法变量

请求语法

GET /v2/variables?type=<Type>
Authorization: Pandora <auth>

响应报文

 {
   "variables": [
     {
        "name": <VariableName>,
        "type": <Type>,
        "value": <Value>,
        "format": <Format>
     },
     ...
   ]
 }

!> 注意：请求url中type参数仅支持system和user, 分别表示列举出系统内置变量和用户本身以创建变量。

根据名称删除魔法变量

请求语法

DELETE /v2/variables/<VariableName>
Authorization: Pandora <auth>

错误代码及相关说明