Simple Log Service:OSSデータインポートジョブの作成

前提条件

• Simple Log Serviceが有効化されています。

• Pythonがインストールされ (Pythonのインストールを参照) 、Python開発環境が設定され (PyCharmを参照) 、Pythonパッケージ管理ツールがインストールされます (pipを参照) 。

  • 新バージョンのSimple Log Service SDK for Pythonは、Python 3.7以降をサポートしています。

  • python-Vコマンドを実行して、インストールされているPythonのバージョンを確認できます。

  • pip3 -Vコマンドを実行して、インストールされているpipのバージョンを確認できます。

• 新バージョンのSimple Log Service SDK for Pythonがインストールされています。

  • CLIで次のコマンドを実行して、新しいバージョンのSimple Log Service SDK for Pythonをインストールします。

    pip install alibabacloud_sls20201230

  • インストールが完了したら、次のコマンドを実行してインストールを確認します。

    pip show alibabacloud_sls20201230

• 環境変数が設定されます。 詳細については、「Linux、macOS、およびWindowsでの環境変数の設定」をご参照ください。

パラメータ説明

def create_ossingestion(
        self,
        project: str,
        request: sls_20201230_models.CreateOSSIngestionRequest,
) -> sls_20201230_models.CreateOSSIngestionResponse:

リクエストパラメーター

リクエストパラメーターの説明:

class CreateOSSIngestionRequest(TeaModel):
    def __init__(
        self,
        configuration: OSSIngestionConfiguration = None,
        description: str = None,
        display_name: str = None,
        name: str = None,
        schedule: Schedule = None,
    ):

スケジュールパラメータの説明:

class Schedule(TeaModel):
    def __init__(
        self,
        cron_expression: str = None,
        delay: int = None,
        interval: str = None,
        run_immediately: bool = None,
        time_zone: str = None,
        type: str = None,
    ):

設定パラメーターの説明

class OSSIngestionConfiguration(TeaModel):
    def __init__(
        self,
        logstore: str = None,
        source: OSSIngestionConfigurationSource = None,
    ):

ソースパラメーターの説明:

class OSSIngestionConfigurationSource(TeaModel):
    def __init__(
        self,
        bucket: str = None,
        compression_codec: str = None,
        encoding: str = None,
        end_time: int = None,
        endpoint: str = None,
        format: Dict[str, Any] = None,
        interval: str = None,
        pattern: str = None,
        prefix: str = None,
        restore_object_enabled: bool = None,
        role_arn: str = None,
        start_time: int = None,
        time_field: str = None,
        time_format: str = None,
        time_pattern: str = None,
        time_zone: str = None,
        use_meta_index: bool = None,
    ):

サンプルコード

import os

from alibabacloud_sls20201230.client import Client as Sls20201230Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_sls20201230 import models as sls_20201230_models
from alibabacloud_tea_util.client import Client as UtilClient


def main():
    config = open_api_models.Config(
        # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is configured.
        access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
        # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is configured.
        access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
    # The Simple Log Service endpoint.
    config.endpoint = "cn-hangzhou.log.aliyuncs.com"
    project_name = "ali-test-project"
    client = Sls20201230Client(config)

    # Create an instance of OSSIngestionConfigurationSource.
    source = sls_20201230_models.OSSIngestionConfigurationSource(
        bucket="oss-test",
        compression_codec="none",
        encoding="UTF-8",
        end_time=None,
        endpoint="oss-cn-hangzhou-internal.aliyuncs.com",
        format={"type": "Line"},
        interval="30s",
        use_meta_index=False,
    )
    # Create an instance of OSSIngestionConfiguration.
    oss_config = sls_20201230_models.OSSIngestionConfiguration(
        logstore="test-logstore",
        source=source
    )

    # Create an instance of CreateOSSIngestionRequest.
    create_ossingestions_request = sls_20201230_models.CreateOSSIngestionRequest(
        configuration=oss_config,
        description="Create an OSS data import job",
        display_name="OSS import test",
        name="ingest-oss-123456"
    )
    try:
        create_ossingestions_response = client.create_ossingestion(project_name, create_ossingestions_request)
        print(create_ossingestions_response)
    except Exception as error:
        print(error.message)
        print(error.data.get("Recommend"))
        UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    main()