DataWorks support for LogHub (SLS) data synchronization-DataWorks(DataWorks)-阿里云帮助中心

The LogHub (SLS) data source provides bidirectional data synchronization between Simple Log Service (SLS) and other data sources in DataWorks.

Features

Simple Log Service (SLS) supports the following data synchronization scenarios:

Data synchronization between LogHub and data sources such as MaxCompute across different regions.
Data synchronization between LogHub and data sources such as MaxCompute across different Alibaba Cloud accounts.
Data synchronization between LogHub and data sources such as MaxCompute within the same Alibaba Cloud account.
Data synchronization between LogHub and data sources such as MaxCompute across public cloud and China Finance Cloud accounts.

Limitation

When Data Integration performs an offline write to LogHub (SLS), data duplication may occur if a task is rerun after a failover. This is because LogHub (SLS) is not idempotent.

Supported field types

Data Integration supports the following LogHub (SLS) field types for reading and writing.

Field type	Offline read (LogHub (SLS) Reader)	Offline write (LogHub (SLS) Writer)	Real-time read
STRING	Supported	Supported	Supported

Details:

When writing data to LogHub (SLS) in offline mode
All data types are converted to STRING before being written to LogHub (SLS). The following table lists the data type conversions.
Supported Data Integration internal type
Data type when written to LogHub (SLS)
LONG
STRING
DOUBLE
STRING
STRING
STRING
DATE
STRING
BOOLEAN
STRING
BYTES
STRING

When reading data from LogHub (SLS) in real-time mode

The following metadata fields are automatically included.

LogHub (SLS) real-time synchronization field	Data type	Description
__time__	STRING	SLS reserved field: __time__ specifies the log time when data is written, in UNIX timestamp format (unit: seconds).
__source__	STRING	SLS reserved field: __source__ specifies the device from which logs are collected.
__topic__	STRING	SLS reserved field: __topic__ specifies the topic name.
__tag__:__receive_time__	STRING	The time when the log arrives at the server. After the public IP recording feature is enabled, the server appends this field to the raw log upon receiving it. The value is a UNIX timestamp in seconds.
__tag__:__client_ip__	STRING	The public IP address of the device from which logs are collected. After the public IP recording feature is enabled, the server appends this field to the raw log upon receiving it.
__tag__:__path__	STRING	The path of the log file collected by Logtail. Logtail automatically appends this field to logs.
__tag__:__hostname__	STRING	The hostname of the machine from which Logtail collects data. Logtail automatically appends this field to logs.

Create a data source

Configure a data source

Before you develop a data synchronization task, create a data source in DataWorks. For the procedure, see Create a data source. For detailed descriptions of configuration parameters, view the tooltips for each parameter on the configuration page.

Create a data source across accounts

The following example shows how Account B configures a synchronization task in Data Integration to synchronize LogHub data from Account A to MaxCompute of Account B. Notes for cross-account scenarios:

Use the AccessKey ID and AccessKey Secret of Account A to create a LogHub data source.
In this case, Account B can synchronize data from all SLS projects under Account A.
Use the AccessKey ID and AccessKey Secret of RAM user A1 under Account A to create a LogHub data source.
- Account A grants RAM user A1 general SLS permissions: AliyunLogFullAccess and AliyunLogReadOnlyAccess. For more information, see Grant permissions to a RAM user.
  Note
  After you attach the AliyunLogFullAccess and AliyunLogReadOnlyAccess system policies to a RAM user, the RAM user can access all SLS resources under the Alibaba Cloud account.
- Account A grants RAM user A1 a custom permission on SLS.
  Log on to the RAM console > Permission Management > Permission Policy page with Account A, and click Create Permission Policy.
  For more information about authorization, see Access control overview and RAM custom authorization examples.
  With the following policy, Account B can use RAM user A1 to synchronize data only from SLS projects project_name1 and project_name2.
```
{
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "log:Get*",
                "log:List*",
                "log:CreateConsumerGroup",
                "log:UpdateConsumerGroup",
                "log:DeleteConsumerGroup",
                "log:ListConsumerGroup",
                "log:ConsumerGroupUpdateCheckPoint",
                "log:ConsumerGroupHeartBeat",
                "log:GetConsumerGroupCheckPoint"
            ],
            "Resource": [
                "acs:log:*:*:project/project_name1",
                "acs:log:*:*:project/project_name1/*",
                "acs:log:*:*:project/project_name2",
                "acs:log:*:*:project/project_name2/*"
            ],
            "Effect": "Allow"
        }
    ]
}
```

Data synchronization task development

For information about the entry point for and the procedure of configuring a synchronization task, see the following configuration guides.

Note

When LogHub is used as the source, you can filter data in LogHub by using LogHub query syntax or SPL statements (SLS Processing Language, the syntax used by SLS to process logs) during task configuration. For more information about the syntax, see Appendix 1: Script demo and parameter description.

Configuration guide for single-table offline sync tasks

For the procedure, see Codeless UI configuration and Script mode configuration.
Note
When you configure a sync node in codeless UI mode, the parameter format must be consistent with the parameter configuration format in Appendix 1: Script demos and parameter description.
For the complete parameters and script demo in script mode, see Appendix 1: Script demos and parameter description below.

Configuration guide for single-table real-time sync tasks

For the procedure, see Single-table real-time synchronization task configuration.

Configuration guide for full-database real-time sync and other full-database-level sync tasks

For the procedure, see Configure real-time full-database synchronization tasks.

FAQ

For more common questions, see Data Integration FAQ.

Appendix 1: Script demo and parameter description

Configure a batch synchronization task by using the code editor

If you want to configure a batch synchronization task by using the code editor, you must configure the related parameters in the script based on the unified script format requirements. For more information, see Script mode configuration. The following information describes the parameters that you must configure for data sources when you configure a batch synchronization task by using the code editor.

Reader script demo

Read from a Logstore demo

{
 "type":"job",
 "version":"2.0",// The version number.
 "steps":[
     {
         "stepType":"LogHub",// The plug-in name.
         "parameter":{
             "datasource":"",// The data source.
             "column":[// The columns.
                 "col0",
                 "col1",
                 "col2",
                 "col3",
                 "col4",
                 "__category__",
                 "__source__",
                 "__topic__", // The log topic.
                 "__machineUUID__", // The unique identifier of the collection machine.
                 "__tag__:__hostname__", // The hostname.
                 "__tag__:__path__", // The path.
                 "__time__" // The event time.
             ],
             "beginDateTime":"",// The start time for data consumption.
             "batchSize":"",// The number of data entries retrieved from SLS in a single query.
             "endDateTime":"",// The end time for data consumption.
             "fieldDelimiter":",",// The column delimiter.
             "logstore":""// The name of the destination Logstore.
         },
         "name":"Reader",
         "category":"reader"
     },
     {
         "stepType":"stream",
         "parameter":{},
         "name":"Writer",
         "category":"writer"
     }
 ],
 "setting":{
     "errorLimit":{
         "record":"0"// The number of error records.
     },
     "speed":{
         "throttle":true,// If throttle is set to false, the mbps parameter does not take effect and bandwidth is not throttled. If throttle is set to true, bandwidth is throttled.
            "concurrent":1 // The number of concurrent threads.
            "mbps":"12",// The throttling rate. 1 mbps = 1 MB/s.
     }
 },
 "order":{
     "hops":[
         {
             "from":"Reader",
             "to":"Writer"
         }
     ]
 }
}

Read from a Logstore after SPL data processing demo

{
 "type":"job",
 "version":"2.0",// The version number.
 "steps":[
     {
         "stepType":"LogHub",// The plug-in name.
         "parameter":{
             "datasource":"",// The data source.
             "column":[// The columns.
                 "col100",
                 "col1"
             ],
             "beginDateTime":"",// The start time for data consumption.
             "batchSize":"",// The number of data entries retrieved from SLS in a single query.
             "endDateTime":"",// The end time for data consumption.
             "fieldDelimiter":",",// The column delimiter.
             "logstore":"",// The name of the destination Logstore.
             "query":"* | where regexp_like(col0, '[0-9]+') | project col100=col0, col1 " // SPL
         },
         "name":"Reader",
         "category":"reader"
     },
     {
         "stepType":"stream",
         "parameter":{},
         "name":"Writer",
         "category":"writer"
     }
 ],
 "setting":{
     "errorLimit":{
         "record":"0"// The number of error records.
     },
     "speed":{
         "throttle":true,// If throttle is set to false, the mbps parameter does not take effect and bandwidth is not throttled. If throttle is set to true, bandwidth is throttled.
            "concurrent":1 // The number of concurrent threads.
            "mbps":"12",// The throttling rate. 1 mbps = 1 MB/s.
     }
 },
 "order":{
     "hops":[
         {
             "from":"Reader",
             "to":"Writer"
         }
     ]
 }
}

Read from a Logstore after extending fields with SPL demo

{
 "type":"job",
 "version":"2.0",// The version number.
 "steps":[
     {
         "stepType":"LogHub",// The plug-in name.
         "parameter":{
             "datasource":"",// The data source.
             "column":[// The columns.
               "col0",
               "col1",
               "col2",
               "col3",
               "col4",
               "__category__",
               "__source__",
               "__topic__", // The log topic.
               "__machineUUID__", // The unique identifier of the collection machine.
               "__tag__:__hostname__", // The hostname.
               "__tag__:__path__", // The path.
               "__time__", // The event time.
               "col100",
               "col101"
             ],
             "beginDateTime":"",// The start time for data consumption.
             "batchSize":"",// The number of data entries retrieved from SLS in a single query.
             "endDateTime":"",// The end time for data consumption.
             "fieldDelimiter":",",// The column delimiter.
             "logstore":"",// The name of the destination Logstore.
             "query":"* | where regexp_like(col0, '[0-9]+') | extend col100=cast(col2 as BIGINT), extend col101=date_parse(col3, '%Y-%m-%d %H:%i') ",// SPL
         },
         "name":"Reader",
         "category":"reader"
     },
     {
         "stepType":"stream",
         "parameter":{},
         "name":"Writer",
         "category":"writer"
     }
 ],
 "setting":{
     "errorLimit":{
         "record":"0"// The number of error records.
     },
     "speed":{
         "throttle":true,// If throttle is set to false, the mbps parameter does not take effect and bandwidth is not throttled. If throttle is set to true, bandwidth is throttled.
            "concurrent":1 // The number of concurrent threads.
            "mbps":"12",// The throttling rate. 1 mbps = 1 MB/s.
     }
 },
 "order":{
     "hops":[
         {
             "from":"Reader",
             "to":"Writer"
         }
     ]
 }
}

Reader script parameters

Parameter	Description	Required	Default value
endPoint	The SLS endpoint is the URL for accessing a project and its log data. It is determined by the Alibaba Cloud region where the project resides. For the endpoint of each region, see Endpoints.	Yes	None
accessId	The AccessKey ID for accessing SLS.	Yes	None
accessKey	The AccessKey secret for accessing SLS.	Yes	N/A
project	The name of the SLS project. A project is the resource management unit in SLS for isolating and controlling resources.	Yes	N/A
logstore	The name of the Logstore. A Logstore is the unit in SLS for collecting, storing, and querying log data.	Yes	N/A
batchSize	The number of data records to retrieve from SLS per query.	No	128
column	The column names in each data record. SLS metadata fields such as log topic, collection machine UUID, hostname, path, and log time can also be configured as synchronization columns. Note Column names are case-sensitive. For more information, see the description of configurable values in Appendix 3: Configurable values for column below.	Yes	N/A
beginDateTime	The start time for data consumption, which is the time when log data arrives at LogHub (SLS). This parameter defines the left boundary (inclusive) of the time range. The value is a time string in yyyyMMddHHmmss format (for example, 20180111013000) and can be used with DataWorks scheduling parameters. For example, in the Scheduling Settings on the right side of the node editing page, configure `beginDateTime=${yyyymmdd-1}` in the Parameter field. Then, set the Log Start Time to ${beginDateTime}000000, which indicates that the log start time is 00:00:00 of the business date. For more information, see Scheduling parameters. Note beginDateTime and endDateTime must be used together. To synchronize all data, you can set beginDateTime to the start time of the data and endDateTime to the current day. However, this may consume a large amount of resources due to the large data volume. Adjust the resource group specifications as needed.	Yes	N/A
endDateTime	The end time for data consumption, which defines the right boundary (exclusive) of the time range. The value is a time string in yyyyMMddHHmmss format (for example, 20180111013010) and can be used with DataWorks scheduling parameters. For example, in the Scheduling Settings on the right side of the node editing page, configure endDateTime=${yyyymmdd} in the Parameter field. Then, set the Log End Time to ${endDateTime}000000, which indicates that the log end time is 00:00:00 of the day after the business date. For more information, see Scheduling parameters. Important The time specified for endDatetime must be earlier than `（2038-01-19 11:14:07 +8:00）`. Otherwise, data may fail to be pulled. The endDateTime of the previous cycle must be equal to or later than the beginDateTime of the next cycle. Otherwise, data in some ranges may fail to be pulled.	Yes	N/A
query	Filters data in LogHub by using LogHub query syntax or SPL statements (SLS Processing Language is the syntax used by SLS to process logs).	Yes	N/A

Note

If data is missing when you read from LogHub for synchronization, check in the LogHub console whether the receive_time metadata field of the data falls within the time range configured for the task.

Writer script demo

{
    "type": "job",
    "version": "2.0",// The version number.
    "steps": [
        { 
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "LogHub",// The plug-in name.
            "parameter": {
                "datasource": "",// The data source.
                "column": [// The columns.
                    "col0",
                    "col1",
                    "col2",
                    "col3",
                    "col4",
                    "col5"
                ],
                "topic": "",// The selected topic.
                "batchSize": "1024",// The number of records to submit at a time.
                "logstore": ""// The name of the destination LogService Logstore.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""// The number of error records.
        },
        "speed": {
            "throttle":true,// If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
            "concurrent":3, // The maximum number of concurrent threads.
            "mbps":"12"// The throttling rate. 1 mbps = 1 MB/s.
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Writer script parameters

Note

LogHub (SLS) Writer receives data from the reader through the Data Integration framework, converts all data types to STRING, and pushes the data to LogHub (SLS) in a batch by using the LogService Java SDK when the specified batchSize is reached.

Parameter	Description	Required	Default value
endpoint	The SLS endpoint is the URL for accessing a project and its log data. It is determined by the Alibaba Cloud region where the project resides. For service endpoints of each region, see Endpoints.	Yes	None
accessKeyId	The AccessKeyId used to access Simple Log Service.	Yes	None
accessKeySecret	The AccessKeySecret used to access Simple Log Service.	Yes	None
project	The name of the SLS project.	Yes	None
logstore	The name of the Logstore. A Logstore is the unit for collecting, storing, and querying log data in SLS.	Yes	None
topic	The topic name for the SLS destination.	No	Empty string
batchSize	The number of data records to synchronize per batch. The default value is 1,024, and the maximum value is 4,096. Note Each batch must not exceed 5 MB. Adjust the number of records per batch based on the size of individual records.	No	1,024
column	The column name in each data record.	Yes	None

Appendix 2: LogHub SPL syntax filtering

When LogHub is the data source, you can filter data by using LogHub query syntax or SPL statements (SLS Processing Language) during task configuration. The syntax is described as follows:

Note

For more information about SPL, see SPL syntax.

Scenario	SQL statement	SPL statement
Data filtering	`SELECT * WHERE Type='write'`	Conditional filtering. `* \| where Type='write'` Fuzzy query. `* \| where Type like '%write%'` Regular expression. `* \| where regexp_like(server_protocol, '\d+')` More (SQL expressions). `* \| where <sql-expr>`
Field processing and selection	Select specific fields and rename them: `SELECT "__tag__:node" AS node, path`	Select specific fields and rename them. `* \| project node="__tag__:node", path` Select fields by pattern. `* \| project -wildcard "__tag__:"` Rename specific fields without affecting other fields. ` \| project-rename node="__tag__:node"` Exclude fields by pattern. `* \| project-away -wildcard "__tag__:*"`
Data transformation (call SQL functions)	Convert data types, parse timestamps, and more: `SELECT CAST(Status AS BIGINT) AS Status, date_parse(Time, '%Y-%m-%d %H:%i') AS Time`	Convert data types, parse timestamps, and more: `* \| extend Status=cast(Status as BIGINT), extend Time=date_parse(Time, '%Y-%m-%d %H:%i')`
Field extraction	Regex extraction: `SELECT CAST(Status AS BIGINT) AS Status, date_parse(Time, '%Y-%m-%d %H:%i') AS Time` JSON extraction: `SELECT CAST(Status AS BIGINT) AS Status, date_parse(Time, '%Y-%m-%d %H:%i') AS Time`	Regex extraction: One-time matching. `* \| parse-regexp protocol, '(\w+)/(\d+)' as scheme, version` JSON extraction: Full expansion. `* \| parse-json -path='$.0' content` CSV extraction. `* \| parse-csv -delim='^_^' content as ip, time, host`

Appendix 3: Column configuration

The full-field read logic is triggered only when column contains a single element that is * or "*". Otherwise, each column element is processed in the normal matching order. For example, if the configuration is ["*", "__time__"], * does not trigger the full-field read logic. Instead, it is matched as a regular field name, which typically outputs null.

The normal matching order is: Regular log content field → Reader built-in meta field mapping → Single-quoted constant → __tag__:<tagKey> → null. If a regular log content field has the same name as a meta field or LogTag key, the regular log content field takes precedence.

The following table lists commonly used fields recommended for user-facing configuration in the LogHub Reader column setting, based on the field definitions of SLS reserved fields. Whether a field has a value depends on the actual Log content, LogGroup metadata, and LogTag retrieved.

Configuration method	Behavior
Regular field name, such as `content` or `level`	Reads from the content fields (contents) of each log entry by key. Field names are case-sensitive.
`__time__`, `__logtime__`	Reads `log.getTime()` for each log entry, which is the log time as a second-level timestamp.
`__source__`	Reads the `source` field of the LogGroup, which corresponds to the log source device.
`__topic__`	Reads the `topic` field of the LogGroup, which corresponds to the log topic.
`__category__`	Reads the `category` field of the LogGroup.
`__machineUUID__`	Reads the `machineUUID` field of the LogGroup.
`__tag__:__receive_time__`	Reads the `__receive_time__` field from LogTag, which indicates the time when the log arrives at the server. This tag must actually exist in the log.
`__tag__:__client_ip__`	Reads the `__client_ip__` field from LogTag, which indicates the public IP address of the log source device. This tag must actually exist in the log.
`__tag__:__path__`	Reads the `__path__` field from LogTag, which typically indicates the log file path collected by Logtail. This tag must actually exist in the log.
`__tag__:__hostname__`	Reads the `__hostname__` field from LogTag, which typically indicates the hostname of the machine from which Logtail collects data. This tag must actually exist in the log.
`__tag__:<tagKey>`	Reads other specified LogTags, such as `__tag__:__pack_id__` and `__tag__:__user_defined_id__`.
`__raw_log__`, `__raw__`	If this field exists as a log content field, it can be read as a regular field name.
`'Fixed value'`	Outputs a constant string (the content after removing the outer single quotes).
`["*"]`	Takes effect only when `column` contains only a single `*`. Reads all regular log content fields, sorts them by key, and outputs each column in the `key:value` format.
Unmatched fields	Outputs `null` without reporting an error.

Configuration example:

"column": [
  "content",
  "__time__",
  "__source__",
  "__tag__:__receive_time__",
  "'constant_value'"
]

Supported Data Integration internal type	Data type when written to LogHub (SLS)
LONG	STRING
DOUBLE	STRING
STRING	STRING
DATE	STRING
BOOLEAN	STRING
BYTES	STRING