Python SDK Getting Started

1. Prepare the job

This job counts the occurrences of "INFO", "WARN", "ERROR", and "DEBUG" in a log file.

The job includes three tasks: split, count, and merge.

• The split task divides the log file into three parts.

• The count task counts the occurrences of "INFO", "WARN", "ERROR", and "DEBUG" in each part. The InstanceCount for the count task must be set to 3. This starts three instances to run the count program simultaneously.

• The merge task combines the results from the count tasks.

(1) Upload the data file to OSS

Download the data file for this example: log-count-data.txt

Upload log-count-data.txt to: oss://your-bucket/log-count/log-count-data.txt

• Replace your-bucket with the name of the bucket that you created. This example assumes that the bucket is in the China (Shenzhen) region.

• For more information about how to upload files to Object Storage Service (OSS), see Upload files.

(2) Upload the task program to OSS

This example uses a job program written in Python. Download the program for this example: log-count.tar.gz

You do not need to change the sample code. Upload the log-count.tar.gz file directly to an OSS path, such as: oss://your-bucket/log-count/log-count.tar.gz

The upload method is described in the previous section.

• BatchCompute supports only compressed packages with the .tar.gz extension. You must package the files using gzip. Otherwise, the package cannot be parsed.

• If you want to modify the code, decompress the package, make your changes, and then re-package the files.

  Run the following command:

  > cd log-count  #Enter the folder
  > tar -czf log-count.tar.gz * #Package all files in this folder into log-count.tar.gz

  Run this command to view the contents of the compressed package:

  $ tar -tvf log-count.tar.gz

  The following list appears:

  conf.py
  count.py
  merge.py
  split.py

2. Create (submit) the job using the SDK

For more information about how to download and install the Python SDK, see Download and install the Python SDK.

When you use API version 20151111, you must specify a cluster ID or use anonymous cluster parameters to submit a job. This example uses an anonymous cluster. You must configure two parameters for an anonymous cluster:

• An available image ID. You can use a system-provided image or create a custom image. For more information, see Custom images.

• The instance type. For more information, see Supported instance types.

In OSS, create a path to store the program output (StdoutRedirectPath) and error logs (StderrRedirectPath). In this example, the path is oss://your-bucket/log-count/logs/

To run this example, you must modify the variables in the program comments to use your actual OSS paths.

The following code provides a template for submitting a job using the Python SDK. For descriptions of the parameters in the program, see Parameter description.

#encoding=utf-8
import sys

from batchcompute import Client, ClientError
from batchcompute import CN_SHENZHEN as REGION    #Set the region as needed.
from batchcompute.resources import (
    JobDescription, TaskDescription, DAG, AutoCluster, Configs, Networks, VPC,
)

ACCESS_KEY_ID='' # Enter your AccessKey ID.
ACCESS_KEY_SECRET='' # Enter your AccessKey secret.

IMAGE_ID = 'img-ubuntu' #Enter your image ID here.
INSTANCE_TYPE = 'ecs.sn1.medium' # Enter an instance type that is supported in the region.
WORKER_PATH = '' # 'oss://your-bucket/log-count/log-count.tar.gz'  Enter the OSS path where you uploaded log-count.tar.gz.
LOG_PATH = '' # 'oss://your-bucket/log-count/logs/' Enter the OSS path that you created for error feedback and task outputs.
OSS_MOUNT= '' # 'oss://your-bucket/log-count/' Mount to /home/inputs and /home/outputs at the same time.

client = Client(REGION, ACCESS_KEY_ID, ACCESS_KEY_SECRET)

def main():
    try:
        job_desc = JobDescription()

        # Create auto cluster.
        cluster = AutoCluster()
        cluster.InstanceType = INSTANCE_TYPE
        cluster.ResourceType = "OnDemand"
        cluster.ImageId = IMAGE_ID
        configs = Configs()
        networks  = Networks()
        vpc = VPC()
        vpc.CidrBlock = '192.168.0.0/16'
        # vpc.VpcId = "vpc-8vbfxdyhxxxx"
        networks.VPC = vpc
        configs.Networks = networks
        # Set the system disk type (cloud_efficiency or cloud_ssd) and size (in GB).
        configs.add_system_disk(size=40, type_='cloud_efficiency')
        configs.InstanceCount = 1
        cluster.Configs = configs

        # Create split task.
        split_task = TaskDescription()
        split_task.Parameters.Command.CommandLine = "python split.py"
        split_task.Parameters.Command.PackagePath = WORKER_PATH
        split_task.Parameters.StdoutRedirectPath = LOG_PATH
        split_task.Parameters.StderrRedirectPath = LOG_PATH
        split_task.InstanceCount = 1
        split_task.AutoCluster = cluster
        split_task.InputMapping[OSS_MOUNT]='/home/input'
        split_task.OutputMapping['/home/output'] = OSS_MOUNT


        # Create map task.
        count_task = TaskDescription(split_task)
        count_task.Parameters.Command.CommandLine = "python count.py"
        count_task.InstanceCount = 3
        count_task.InputMapping[OSS_MOUNT] = '/home/input'
        count_task.OutputMapping['/home/output'] = OSS_MOUNT

        # Create merge task
        merge_task = TaskDescription(split_task)
        merge_task.Parameters.Command.CommandLine = "python merge.py"
        merge_task.InstanceCount = 1
        merge_task.InputMapping[OSS_MOUNT] = '/home/input'
        merge_task.OutputMapping['/home/output'] = OSS_MOUNT

        # Create task dag.
        task_dag = DAG()
        task_dag.add_task(task_name="split", task=split_task)
        task_dag.add_task(task_name="count", task=count_task)
        task_dag.add_task(task_name="merge", task=merge_task)
        task_dag.Dependencies = {
            'split': ['count'],
            'count': ['merge']
        }

        # Create job description.
        job_desc.DAG = task_dag
        job_desc.Priority = 99 # 0-1000
        job_desc.Name = "log-count"
        job_desc.Description = "PythonSDKDemo"
        job_desc.JobFailOnInstanceFail = True

        job_id = client.create_job(job_desc).Id
        print('job created: %s' % job_id)

    except ClientError, e:
        print (e.get_status_code(), e.get_code(), e.get_requestid(), e.get_msg())

if __name__ == '__main__':
    sys.exit(main())

3. View the job status

Use the Get Job method in the SDK to retrieve the job status. For more information, see Get job information.

jobInfo = client.get_job(job_id)
print (jobInfo.State)

The job can be in one of the following states: Waiting, Running, Finished, Failed, or Stopped.

4. View the results

Log on to the OSS console to view the /log-count/merge_result.json file in your bucket.

The content should be as follows:

{"INFO": 2460, "WARN": 2448, "DEBUG": 2509, "ERROR": 2583}

You can also use the OSS SDK to retrieve the results.