Simple Log Service:Create a scheduled SQL job

Prerequisites

Parameters

def create_scheduled_sql(self, project_name, scheduled_sql):

Request parameters

scheduled_sql parameter

schedule parameter

Configuration parameter

{
 addLabels: "{}",
 hashLabels: "[]",
 labelKeys: "[\"your label1\",\"your label2\"]",
 metricKeys: "[\"your Indicator1\",\"your Indicator2\"]",
 metricName: "",
 timeKey: ""
}

Response parameters

See CreateScheduledSQL.

Sample code

import os
import time
from aliyun.log import LogClient
from aliyun.log.scheduled_sql import *

accessKeyId = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')  # The AccessKey ID.
accessKeySecret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')  # The AccessKey secret.

endpoint = "cn-hangzhou.log.aliyuncs.com"  # The endpoint of the region where the source project is located.
roleArn = "acs:ram::141******5616316:role/aliyunserviceroleforslsaudit"  # The ARN of the role.
project = "demo-test-project"  # The name of the source project.
source_logstore = "test-logstore"  # The name of the source Logstore.
source_metricstore = ""  # The name of the source Metricstore.

dest_endpoint = "cn-hangzhou.log.aliyuncs.com"  # The endpoint of the region where the destination project is located.
dest_role_arn = "acs:ram::141******5616316:role/aliyunserviceroleforslsaudit"  # The ARN of the destination role.
dest_project = "demo-test-project"  # The name of the destination project.
dest_logstore = "test-logstore2"  # The name of the destination Logstore.
dest_metricstore = ""  # The name of the destination Metricstore.

from_time = int(time.time()) - 360  # The start time of the scheduled SQL job.

job_name = "test-001"  # The job name.
display_name = "test-001"  # The display name.
description = "Create a scheduled SQL job"  # The description.

script = "* | select *"  # The SQL script.
promql = "* | select promql_query_range('key{}') from metrics limit 1000"  # The PromQL script.

instance_id = ""  # The job instance ID for the scheduled SQL job.
delay_seconds = 0  # The delay for the scheduled SQL job, in seconds.

# Three possible values for the variable data_format: "log2log", "log2metric", and "metric2metric".
data_format = "log2log"

# Possible values for the variable schedule_type: "FixedRate", "Daily", "Weekly", "Hourly", and "Cron".
# schedule_type = "FixedRate"
schedule_type = "FixedRate"

client = LogClient(endpoint, accessKeyId, accessKeySecret)


def generate_schedule_sql():
    schedule_rule = generate_schedule_rule()
    config = generate_config()

    schedule_sql = ScheduledSQL()
    schedule_sql.setName(job_name)
    schedule_sql.setDisplayName(display_name)
    schedule_sql.setDescription(description)
    schedule_sql.setSchedule(schedule_rule)
    schedule_sql.setConfiguration(config)
    return schedule_sql


def generate_schedule_rule():
    schedule_rule = JobSchedule()
    schedule_rule.setType(schedule_type)
    schedule_rule.setDelay(delay_seconds)
    schedule_rule.setTimeZone('+0800')
    if schedule_type == "FixedRate":
        # The interval value must be a string in the format of "Xs", "Xm", "Xh", or "Xd",
        # where X is an integer that indicates the number of seconds, minutes, hours, or days.
        schedule_rule.setInterval('60s')
    elif schedule_type == "Cron":
        # The cron expression must be a valid cron string.
        schedule_rule.setCronExpression("*/2 * * * *")
    elif schedule_type == "Daily":
        # The hour value must be an integer from 0 to 23.
        schedule_rule.setHour(1)
    elif schedule_type == "Weekly":
        # The day of the week value must be an integer from 1 to 7.
        # The hour value must be an integer from 0 to 23.
        schedule_rule.setDayOfWeek(1)
        schedule_rule.setHour(2)
    elif schedule_type == "Hourly":
        pass

    return schedule_rule


def generate_config():
    query = script
    if data_format == 'log2log':
        source = source_logstore
        dest = dest_logstore
    elif data_format == 'log2metric':
        source = source_logstore
        dest = dest_metricstore
    elif data_format == 'metric2metric':
        source = source_metricstore
        dest = dest_metricstore
        query = promql

    parameters = generate_log2log_params()
    config = ScheduledSQLConfiguration()
    config.setScript(query)
    config.setSqlType('searchQuery')
    config.setDestEndpoint(dest_endpoint)
    config.setDestProject(dest_project)
    config.setSourceLogstore(source)
    config.setDestLogstore(dest)
    config.setRoleArn(roleArn)
    config.setDestRoleArn(dest_role_arn)
    config.setFromTimeExpr("@m - 5m")
    config.setToTimeExpr("@m")
    config.setMaxRunTimeInSeconds(1800)
    config.setResourcePool("enhanced")
    config.setMaxRetries(10)
    config.setFromTime(from_time)
    config.setToTime(0)
    config.setDataFormat(data_format)
    config.setParameters(parameters)
    return config


def generate_log2log_params():
    parameters = ScheduledSQLBaseParameters()
    return parameters


if __name__ == '__main__':
    schedule_sql = generate_schedule_sql()
    res = client.create_scheduled_sql(dest_project, schedule_sql)
    res.log_print()

Sample response

CreateScheduleSqlResponse:
headers: {'Server': 'AliyunSLS', 'Content-Length': '0', 'Connection': 'keep-alive', 'Access-Control-Allow-Origin': '*', 'Date': 'Fri, 01 Nov 2024 07:04:39 GMT', 'x-log-time': '1730444679', 'x-log-requestid': '67247D87DB04A454D92A6F6B'}

Process finished with exit code 0

References

• For more information about the API operations used to manage scheduled SQL jobs, see the following topics:

  • CreateScheduledSQL

  • DeleteScheduledSQL

  • UpdateScheduledSQL

  • EnableScheduledSQL

  • DisableScheduledSQL

  • ListScheduledSQLs

  • GetScheduledSQL

• For more sample code, see Alibaba Cloud Simple Log Service SDK for Python on GitHub.