DataWorks support for LogHub (SLS) data synchronization - 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 performed by LogHub (SLS) Writer.
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 log data is written. The value is a UNIX timestamp in seconds.
__source__	STRING	SLS reserved field: __source__ specifies the source device of the log.
__topic__	STRING	SLS reserved field: __topic__ specifies the topic name.
__tag__:__receive_time__	STRING	The time when the log arrives at the server. After you enable the feature for recording public IP addresses, the server appends this field to the raw log upon reception. The value is a UNIX timestamp in seconds.
__tag__:__client_ip__	STRING	The public IP address of the source device of the log. After you enable the feature for recording public IP addresses, the server appends this field to the raw log upon reception.
__tag__:__path__	STRING	The path of the log file collected by Logtail. Logtail automatically appends this field to the log.
__tag__:__hostname__	STRING	The hostname of the source machine from which Logtail collects data. Logtail automatically appends this field to the log.

Create a data source

Configure a data source

Before you develop a data synchronization task, you must create a corresponding data source in DataWorks. For the procedure, see Create a data source. For detailed descriptions of configuration parameters, see the tooltips displayed on the configuration page.

Create a data source across Alibaba Cloud accounts

The following example describes how to use Account B to configure a synchronization task in Data Integration to synchronize LogHub data from Account A to MaxCompute of Account B. The special notes for cross-account scenarios are as follows:

Create a LogHub data source by using the AccessKey ID and AccessKey Secret of Account A.
In this case, Account B can synchronize data from all SLS projects under Account A.
Create a LogHub data source by using the AccessKey ID and AccessKey Secret of RAM user A1 under Account A.
- Account A grants RAM user A1 the general permissions on SLS, which are the AliyunLogFullAccess and AliyunLogReadOnlyAccess policies. For more information, see Create a RAM user and grant permissions.
  Note
  After you attach the AliyunLogFullAccess and AliyunLogReadOnlyAccess system policies to a RAM user, the RAM user can query 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 by using Account A, and click Create Permission Policy.
  For information about relevant authorization, see Access control overview and RAM custom authorization examples.
  After you authorize RAM user A1 based on the following policy, Account B can synchronize data only from SLS projects project_name1 and project_name2 through RAM user A1.
```
{
    "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 a LogHub data source is used as the source, you can filter data in LogHub by using the LogHub query syntax or SPL statements (SLS Processing Language, the syntax used by SLS to process logs) during task configuration. For details about the syntax, see Appendix: LogHub data filtering.

Single-table offline synchronization task configuration guide

For the procedure, see Configure a single-table offline synchronization task in the codeless UI and Configure a single-table offline synchronization task by using a script.
Note
When you configure a sync node in codeless UI mode, the parameter format must be consistent with the parameter configuration format in script mode.
For the complete parameters and script demo in script mode, see Script mode parameters below.

Single-table real-time synchronization task configuration guide

For the procedure, see Configure a single-table real-time synchronization task.

Full-database real-time synchronization and other database-level synchronization configuration guide

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

FAQ

For more common issues about Data Integration, 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 queried from SLS at a time.
             "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, which means throttling is disabled. If throttle is set to true, throttling is enabled.
            "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 data processing 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.
                 "col100",
                 "col1"
             ],
             "beginDateTime":"",// The start time for data consumption.
             "batchSize":"",// The number of data entries queried from SLS at a time.
             "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, which means throttling is disabled. If throttle is set to true, throttling is enabled.
            "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 queried from SLS at a time.
             "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, which means throttling is disabled. If throttle is set to true, throttling is enabled.
            "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 Simple Log Service endpoint is the URL used to access a project and its log data. It is related to the Alibaba Cloud region where the project resides and the project name. For the service endpoints of each region, see Endpoints.	Yes	None
accessId	The AccessKey ID used to access Simple Log Service, which identifies the user.	Yes	None
accessKey	The AccessKey used to access Simple Log Service for user authentication.	Yes	None
project	The name of the destination Simple Log Service project. A project is the resource management unit in Simple Log Service and is used to isolate and manage resources.	Yes	None
logstore	The name of the destination Logstore. A Logstore is the unit in Simple Log Service for collecting, storing, and querying log data.	Yes	None
batchSize	The number of data entries retrieved from Simple Log Service in a single query.	No	128
column	The column names in each data entry. You can configure metadata fields of Simple Log Service as synchronization columns. Simple Log Service supports metadata such as log topic, unique identifier of the collection machine, hostname, path, and log time. Note Column names are case-sensitive. For more information, see the configurable column values section below.	Yes	None
beginDateTime	The start time for data consumption, which is the time when log data arrives at LogHub (SLS). This parameter specifies the left boundary (inclusive) of the time range. Set this parameter to a time string in yyyyMMddHHmmss format (for example, 20180111013000). You can use this parameter together with DataWorks scheduling parameters. For example, in Scheduling Settings on the right side of the node editing page, set `beginDateTime=${yyyymmdd-1}` in Parameter, and then set Log Start Time to ${beginDateTime}000000. This indicates that the log start time is 00:00:00 on 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 data volume. Adjust the resource group specifications as needed. The underlying time filter field is fixed to write timestamp (the time when the log arrives at LogHub (SLS)). `beginDateTime` controls only the time range of data consumption. It does not change the underlying filter field and cannot be changed to a custom business field such as `start_time`.	Yes	None
endDateTime	The end time for data consumption, which is the right boundary (exclusive) of the time range. Set this parameter to a time string in yyyyMMddHHmmss format (for example, 20180111013010). You can use this parameter together with DataWorks scheduling parameters. For example, in Scheduling Settings on the right side of the node editing page, set endDateTime=${yyyymmdd} in Parameter, and then set Log End Time to ${endDateTime}000000. This indicates that the log end time is 00:00:00 on 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 retrieved. The endDateTime of the previous scheduling cycle must be the same as or later than the beginDateTime of the next scheduling cycle. Otherwise, data in some time ranges may not be retrieved. The underlying time filter field is fixed to write timestamp (the time when the log arrives at LogHub (SLS)). `endDateTime` controls only the time range of data consumption. It does not change the underlying filter field and cannot be changed to a custom business field such as `start_time`.	Yes	None
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	None

Note

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

Partition by custom business fields

The time filter for LogHub (SLS) offline synchronization tasks is always based on __tag__:__receive_time__. Filtering or partitioning directly by custom business fields such as start_time is not supported. To partition data into MaxCompute by a custom business field, use the following workaround:

Configure an offline synchronization task to synchronize all LogHub (SLS) data to a MaxCompute staging table (without specifying a partition field).
Create a MaxCompute SQL node and run the following statement to write data into the destination partitioned table by a custom business field (such as start_time):
```
INSERT OVERWRITE TABLE target_table PARTITION(start_time)
SELECT col1, col2, start_time
FROM temp_table;
```
Use scheduling parameters to dynamically pass partition values and implement automatic partitioning by day or by hour.

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 bandwidth is not throttled. If throttle is set to true, bandwidth is throttled.
            "concurrent":3, // The 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 obtains data generated by Reader through the Data Integration framework, and then converts all supported data types to STRING one by one. When the specified batchSize is reached, the data is pushed to LogHub (SLS) in a single batch by using the LogService Java SDK.

Parameter	Description	Required	Default value
endpoint	The Log Service endpoint is the URL used to access a project and its log data. It is determined by the Alibaba Cloud region and the project name. For endpoints of each region, see Service endpoints.	Yes	None
accessKeyId	The AccessKeyId used to access Log Service.	Yes	None
accessKeySecret	The AccessKeySecret used to access Log Service.	Yes	None
project	The name of the destination Log Service project.	Yes	None
logstore	The name of the destination Logstore. A Logstore is the unit in Log Service for collecting, storing, and querying log data.	Yes	None
topic	The topic name in the destination SLS.	No	Empty string
batchSize	The number of data records synchronized to LogHub (SLS) at a time. Default value: 1,024. Maximum value: 4,096. Note The data size for a single batch write to LogHub (SLS) must not exceed 5 MB. Adjust the number of records pushed per batch based on your individual record size.	No	1,024
column	The column name in each data record.	Yes	None

Appendix 2: LogHub SPL syntax filtering

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

Note

For more information about SPL, see SPL overview.

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	Regular expression 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`	Regular expression 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: Configurable column values

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

The normal matching order is: normal log content fields → Reader built-in meta field mapping → single-quoted constants → __tag__:<tagKey> → null. If a normal log content field has the same name as a meta field or LogTag key, the normal log content field takes priority.

The following table lists the fields that are commonly used and recommended for user-facing configurations in the LogHub Reader column settings, based on the field definitions of SLS reserved fields. Whether a field has a value depends on the actual log content, LogGroup metadata, and LogTags retrieved.

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