阿里云大数据开发治理平台 DataWorksLogHub（SLS）数据源-云淘科技

使用限制

数据集成离线写LogHub（SLS）时，由于LogHub（SLS）无法实现幂等，FailOver重跑任务时会引起数据重复。

支持的字段类型

数据集成支持读写的LogHub（SLS）字段类型如下。

其中：

• 离线写LogHub（SLS）时

  会将支持同步的各类型数据均转换成STRING类型后写入LogHub（SLS）。LogHub（SLS） Writer针对LogHub（SLS）类型的转换列表，如下所示。

  支持的数据集成内部类型	写入LogHub（SLS）时的数据类型
  LONG	STRING
  DOUBLE	STRING
  STRING	STRING
  DATE	STRING
  BOOLEAN	STRING
  BYTES	STRING

• 实时读LogHub（SLS）时

  会自带以下元数据字段。

  LogHub（SLS）实时同步字段	数据类型	说明
  __time__	STRING	SLS保留字段：__time__写入日志数据时指定的日志时间，unix时间戳，单位为秒。
  __source__	STRING	SLS保留字段：__source__日志来源设备。
  __topic__	STRING	SLS保留字段：__topic__topic名称。
  __tag__:__receive_time__	STRING	日志到达服务端的时间。开启记录外网IP功能后，服务端接收日志时为原始日志追加该字段。unix时间戳，单位为秒。
  __tag__:__client_ip__	STRING	日志来源设备的公网IP。开启记录外网IP功能后，服务端接收日志时为原始日志追加该字段。
  __tag__:__path__	STRING	Logtail采集的日志文件路径，Logtail为日志自动追加该字段。
  __tag__:__hostname__	STRING	Logtail采集数据的来源机器主机名，Logtail为日志自动追加该字段。

数据同步任务开发：LogHub（SLS）同步流程引导

LogHub（SLS）数据同步任务的配置入口和通用配置流程指导可参见下文的配置指导，详细的配置参数解释可在配置界面查看对应参数的文案提示。

创建数据源

在进行数据同步任务开发时，您需要在DataWorks上创建一个对应的数据源，操作流程请参见创建与管理数据源。

单表离线同步任务配置指导

• 操作流程请参见通过向导模式配置离线同步任务、通过脚本模式配置离线同步任务。

• 脚本模式配置的全量参数和脚本Demo请参见下文的附录：LogHub（SLS）脚本Demo与参数说明。

单表实时同步任务配置指导

操作流程请参见配置单表增量数据实时同步、DataStudio侧实时同步任务配置。

整库离线、整库（实时）全增量、整库（实时）分库分表等整库级别同步配置指导

操作流程请参见数据集成侧同步任务配置。

常见问题

• 读取Loghub同步某字段有数据但是同步过来为空

• 读取Loghub同步少数据

• 读取Loghub字段映射时读到的字段不符合预期

更多其他数据集成常见问题请参见数据集成常见问题。

附录：LogHub（SLS）脚本Demo与参数说明

附录：离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置，您需要在任务脚本中按照脚本的统一格式要求编写脚本中的reader参数和writer参数，脚本模式的统一要求请参见通过脚本模式配置离线同步任务，以下为您介绍脚本模式下的数据源的Reader参数和Writer参数的指导详情。

LogHub（SLS） Reader脚本Demo

{
 "type":"job",
 "version":"2.0",//版本号。
 "steps":[
     {
         "stepType":"LogHub",//插件名。
         "parameter":{
             "datasource":"",//数据源。
             "column":[//字段。
                 "col0",
                 "col1",
                 "col2",
                 "col3",
                 "col4",
                 "C_Category",
                 "C_Source",
                 "C_Topic",
                 "C_MachineUUID", //日志主题。
                 "C_HostName", //主机名。
                 "C_Path", //路径。
                 "C_LogTime" //事件时间。
             ],
             "beginDateTime":"",//数据消费的开始时间位点。
             "batchSize":"",//一次从日志服务查询的数据条数。
             "endDateTime":"",//数据消费的结束时间位点。
             "fieldDelimiter":",",//列分隔符。
             "logstore":""//：目标日志库的名字。
         },
         "name":"Reader",
         "category":"reader"
     },
     { 
         "stepType":"stream",
         "parameter":{},
         "name":"Writer",
         "category":"writer"
     }
 ],
 "setting":{
     "errorLimit":{
         "record":"0"//错误记录数。
     },
     "speed":{
         "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":1 //作业并发数。
            "mbps":"12",//限流，此处1mbps = 1MB/s。
     }
 },
 "order":{
     "hops":[
         {
             "from":"Reader",
             "to":"Writer"
         }
     ]
 }
}

LogHub（SLS） Reader脚本参数

LogHub（SLS） writer脚本Demo

{
    "type": "job",
    "version": "2.0",//版本号。
    "steps": [
        { 
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "LogHub",//插件名。
            "parameter": {
                "datasource": "",//数据源。
                "column": [//字段。
                    "col0",
                    "col1",
                    "col2",
                    "col3",
                    "col4",
                    "col5"
                ],
                "topic": "",//选取topic。
                "batchSize": "1024",//一次性批量提交的记录数大小。
                "logstore": ""//目标LogService LogStore的名称。
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""//错误记录数。
        },
        "speed": {
            "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":3, //作业并发数。
            "mbps":"12"//限流，此处1mbps = 1MB/s。
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

LogHub（SLS） Writer脚本参数

说明

LogHub（SLS） Writer通过数据集成框架获取Reader生成的数据，然后将数据集成支持的类型通过逐一判断转换成STRING类型。当达到您指定的batchSize时，会使用LogService Java SDK一次性推送至LogHub（SLS）。