本文档将介绍如何使用 Python 版 SDK 来提交一个作业,目的是统计一个日志文件中“INFO”,”WARN”,”ERROR”,”DEBUG”出现的次数。

  • 作业准备
    • 上传数据文件到 OSS
    • 上传任务程序到 OSS
  • 使用 SDK 创建(提交)作业
  • 查看结果

1. 作业准备

本作业是统计一个日志文件中“INFO”,”WARN”,”ERROR”,”DEBUG”出现的次数。

该作业包含3个任务: split, count 和 merge:

  • split 任务会把日志文件分成 3 份。
  • count 任务会统计每份日志文件中“INFO”,”WARN”,”ERROR”,”DEBUG”出现的次数 (count 任务需要配置 InstanceCount 为 3,表示同时启动3台机器运行个 count 程序)。
  • merge 任务会把 count 任务的结果统一合并起来。

DAG图例:

DAG

(1) 上传数据文件到OSS

下载本例子所需的数据:log-count-data.txt

将 log-count-data.txt 上传到:oss://your-bucket/log-count/log-count-data.txt

  • your-bucket 表示您自己创建的 bucket,本例子假设 region 为: cn-shenzhen。
  • 如何上传到 OSS,请参考 OSS 上传文档

(2) 上传任务程序到OSS

本例的作业程序是使用 python 编写的, 下载本例子所需的程序: log-count.tar.gz

本例不需要改动示例代码。直接将 log-count.tar.gz 上传到 oss,如上传到:oss://your-bucket/log-count/log-count.tar.gz

如何上传前面已经讲过。

  • BatchCompute 只支持以 tar.gz 为后缀的压缩包, 请注意务必用以上方式(gzip)打包, 否则将会无法解析。
  • 如果您要修改代码,可以解压后修改,然后要用下面的方法打包:

    命令如下:

    > cd log-count  #进入目录
    > tar -czf log-count.tar.gz * #打包,将所有这个目录下的文件打包到 log-count.tar.gz

    可以运行这条命令查看压缩包内容:

    $ tar -tvf log-count.tar.gz

    可以看到以下列表:

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

2. 使用SDK创建(提交)作业

python SDK 的相关下载与安装请参阅 这里

v20151111 版本,提交作业需要指定集群 ID 或者使用匿名集群参数。本例子使用匿名集群方式进行,匿名集群需要配置 2 个参数, 其中:

  • 可用的镜像 ID, 可以使用系统提供的 Image,也可以自行制作镜像, 请参见 镜像制作
  • 实例规格(InstanceType,实例类型),请参考 目前支持类型

在 OSS 中创建存储 StdoutRedirectPath(程序输出结果)和 StderrRedirectPath(错误日志)的文件路径,本例中创建的路径为oss://your-bucket/log-count/logs/

如需运行本例,请按照上文所述的变量获取以及与上文对应的您的 OSS 路径对程序中注释中的变量进行修改。

Python SDK 提交程序模板如下,程序中具体参数含义请参见 参数说明

#encoding=utf-8
import sys

from batchcompute import Client, ClientError
from batchcompute import CN_SHENZHEN as REGION    #这里的region根据实际情况填写
from batchcompute.resources import (
    JobDescription, TaskDescription, DAG, AutoCluster, Configs, Networks, VPC,
)

ACCESS_KEY_ID='' # 填写您的 AK
ACCESS_KEY_SECRET='' # 填写您的 AK

IMAGE_ID = 'img-ubuntu' #这里填写您的镜像 ID
INSTANCE_TYPE = 'ecs.sn1.medium' # 根据实际 region 支持的 InstanceType 填写
WORKER_PATH = '' # 'oss://your-bucket/log-count/log-count.tar.gz'  这里填写您上传的 log-count.tar.gz 的 OSS 存储路径
LOG_PATH = '' # 'oss://your-bucket/log-count/logs/' 这里填写您创建的错误反馈和 task 输出的 OSS 存储路径
OSS_MOUNT= '' # 'oss://your-bucket/log-count/' 同时挂载到/home/inputs 和 /home/outputs

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
        # 设置系统盘type(cloud_efficiency/cloud_ssd)以及size(单位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. 查看作业状态

您可以用 SDK 中的 获取作业信息 方法获取作业状态:

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

State 状态可能为:Waiting, Running, Finished, Failed, Stopped。

4. 查看结果

您可以登录 OSS 控制台 查看 your-bucket 下面的这个文件:/log-count/merge_result.json。

内容应该如下:

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

您也可以使用 OSS 的 SDK 来获取结果。