🦁

AirflowのDAGジョブでPythonSDKを実行する

2022/09/01に公開約7,600字

AirflowでOCIのPythonSDKをインポートする方法の記事でociをインポートすることはできましたが、今度は下記エラーとなりました。

[2022-09-01 06:51:57,215] {taskinstance.py:1481} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1137, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1341, in _execute_task
    result = task_copy.execute(context=context)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/operators/python.py", line 150, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/operators/python.py", line 161, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/oci_list.py", line 68, in object_storage_list
    config = oci.config.from_file()
  File "/home/airflow/.local/lib/python3.6/site-packages/oci/config.py", line 107, in from_file
    raise ConfigFileNotFound("Could not find config file at {}, please follow the instructions in the link to setup the config file https://docs.cloud.oracle.com/en-us/iaas/Content/API/Concepts/sdkconfig.htm".format(expanded_file_location))
oci.exceptions.ConfigFileNotFound: Could not find config file at /home/***/.oci/config, please follow the instructions in the link to setup the config file https://docs.cloud.oracle.com/en-us/iaas/Content/API/Concepts/sdkconfig.htm

エラーの回避

色々試しましたがconfigファイルを読ませる事が出来ず諦めました・・・・
代わりにDAGの中にconfig情報直書きすることで回避させました。

config情報直書き

下記のような感じで記述することで回避しました。

def object_storage_list(**kwargs):
        pem_prefix = '-----BEGIN RSA PRIVATE KEY-----\n'
        pem_suffix = '\n-----END RSA PRIVATE KEY-----'
        key = "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCqUzAq791phsWNj/PSSqGgrCn\
        lKC37wyfFbmeDmsX0yZfuoTgEDU1xup4CuHbB1klu2mW879LQ1uAa2hQuu/QzWh49Tlx30L2yRrSOugXBV\
    ~略~
        5yqMErUTw7Zj+T9XiQZUmcktH6Mo0pKciQKBkOa8jsCrduQvu0VZuixzLo0vPZlU8NcnbLSc="
        """
	key_content = '{}{}{}'.format(pem_prefix, key, pem_suffix)
        config_with_key_content = {
            "user": 'ocid1.user.oc1..aaaaaaaa**************',
            "key_content": key_content,
            "fingerprint": '8b:*****************ce:93',
            "tenancy": 'ocid1.tenancy.oc1..aaaaaaaa***********',
            "region": 'ap-tokyo-1'
        }
	oci.config.validate_config(config_with_key_content)
	object_storage_client = oci.object_storage.ObjectStorageClient(config_with_key_content)
	namespace_name="********"
        bucket_name="test"
        get_object_list = object_storage_client.list_objects(namespace_name,bucket_name)
        return get_object_list.data

実行できたけどまたエラー

[2022-09-01 07:15:30,151] {taskinstance.py:1481} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1137, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1344, in _execute_task
    self.xcom_push(key=XCOM_RETURN_KEY, value=result)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1925, in xcom_push
    session=session,
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/xcom.py", line 79, in set
    value = XCom.serialize_value(value)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/xcom.py", line 226, in serialize_value
    return json.dumps(value).encode('UTF-8')
  File "/usr/local/lib/python3.6/json/__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "/usr/local/lib/python3.6/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/local/lib/python3.6/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/usr/local/lib/python3.6/json/encoder.py", line 180, in default
    o.__class__.__name__)
TypeError: Object of type 'ListObjects' is not JSON serializable

再度調査することに。。。。
このエラーはreturnで返す型がオブジェクトでJSONにパース出来ないのでエラーが起きてるっぽい??
なので無理やり文字列に変換してやる

修正箇所
return str(get_object_list.data[0])
実行結果
*** Reading local file: /opt/airflow/logs/test_oci/object_list/2022-09-01T08:14:01.768048+00:00/1.log
[2022-09-01 08:14:08,303] {taskinstance.py:876} INFO - Dependencies all met for <TaskInstance: test_oci.object_list 2022-09-01T08:14:01.768048+00:00 [queued]>
[2022-09-01 08:14:08,324] {taskinstance.py:876} INFO - Dependencies all met for <TaskInstance: test_oci.object_list 2022-09-01T08:14:01.768048+00:00 [queued]>
[2022-09-01 08:14:08,325] {taskinstance.py:1067} INFO - 
--------------------------------------------------------------------------------
[2022-09-01 08:14:08,325] {taskinstance.py:1068} INFO - Starting attempt 1 of 1
[2022-09-01 08:14:08,327] {taskinstance.py:1069} INFO - 
--------------------------------------------------------------------------------
[2022-09-01 08:14:08,336] {taskinstance.py:1087} INFO - Executing <Task(PythonOperator): object_list> on 2022-09-01T08:14:01.768048+00:00
[2022-09-01 08:14:08,347] {standard_task_runner.py:52} INFO - Started process 65631 to run task
[2022-09-01 08:14:08,352] {standard_task_runner.py:76} INFO - Running: ['***', 'tasks', 'run', 'test_oci', 'object_list', '2022-09-01T08:14:01.768048+00:00', '--job-id', '70', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/oci_list.py', '--cfg-path', '/tmp/tmpv2gue29i', '--error-file', '/tmp/tmpy4xncbn5']
[2022-09-01 08:14:08,352] {standard_task_runner.py:77} INFO - Job 70: Subtask object_list
[2022-09-01 08:14:08,390] {logging_mixin.py:104} INFO - Running <TaskInstance: test_oci.object_list 2022-09-01T08:14:01.768048+00:00 [running]> on host eefb3556693f
[2022-09-01 08:14:08,449] {taskinstance.py:1282} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=test_oci
AIRFLOW_CTX_TASK_ID=object_list
AIRFLOW_CTX_EXECUTION_DATE=2022-09-01T08:14:01.768048+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2022-09-01T08:14:01.768048+00:00
[2022-09-01 08:14:08,547] {python.py:151} INFO - Done. Returned value was: {
  "archival_state": null,
  "etag": null,
  "md5": null,
  "name": "test.txt",
  "size": null,
  "storage_tier": null,
  "time_created": null,
  "time_modified": null
}
[2022-09-01 08:14:08,576] {taskinstance.py:1191} INFO - Marking task as SUCCESS. dag_id=test_oci, task_id=object_list, execution_date=20220901T081401, start_date=20220901T081408, end_date=20220901T081408
[2022-09-01 08:14:08,601] {taskinstance.py:1245} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2022-09-01 08:14:08,641] {local_task_job.py:151} INFO - Task exited with return code 0

Discussion

ログインするとコメントできます