Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

파이프라인 상태 모니터링

이 페이지에서는 파이프라인 상태와 같은 Cloud Data Fusion 파이프라인 이벤트를 Pub/Sub 주제에 게시하는 방법을 설명합니다. 또한 Pub/Sub 메시지를 처리하고 실패한 파이프라인 식별 및 재시도와 같은 작업을 수행하는 Cloud Run Functions를 만드는 방법을 설명합니다.

시작하기 전에

Pub/Sub에서 Cloud Data Fusion 파이프라인 이벤트를 게시할 수 있는 주제를 만듭니다.

필요한 역할

Cloud Data Fusion 서비스 계정에 파이프라인 이벤트를 Pub/Sub 주제에 게시하는 데 필요한 권한이 있는지 확인하려면 관리자에게 Pub/Sub 게시자 (roles/pubsub.publisher) IAM 역할을 Pub/Sub 주제를 만드는 프로젝트에서 Cloud Data Fusion 서비스 계정에 부여해 달라고 요청하세요.

역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

관리자는 커스텀 역할이나 다른 사전 정의된 역할을 통해 Cloud Data Fusion 서비스 계정에 필요한 권한을 부여할 수도 있습니다.

Cloud Data Fusion 인스턴스에서 이벤트 게시 관리

버전 6.7.0 이상에서 REST API를 사용하여 신규 및 기존 Cloud Data Fusion 인스턴스에서 이벤트 게시를 관리할 수 있습니다.

새 인스턴스에서 이벤트 게시

새 인스턴스를 만들고 EventPublishConfig 필드를 포함합니다. 새 인스턴스에 필요한 필드에 대한 자세한 내용은 인스턴스 리소스 참조를 확인하세요.

curl -X POST \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  "https://datafusion.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/instances?instanceId=INSTANCE_ID" \
  -d '{
    "version": "VERSION_NUMBER",
    "event_publish_config": {
      "enabled": true,
      "topic": "projects/PROJECT_ID/topics/TOPIC_ID"
    }
  }'

다음을 바꿉니다.

PROJECT_ID: Google Cloud 프로젝트 ID입니다.
LOCATION: 프로젝트의 위치
INSTANCE_ID: Cloud Data Fusion 인스턴스의 ID
VERSION_NUMBER: 인스턴스를 만드는 Cloud Data Fusion의 버전(예: 6.10.1)
TOPIC_ID: Pub/Sub 주제의 ID

기존 Cloud Data Fusion 인스턴스에서 이벤트 게시 사용 설정

기존 Cloud Data Fusion 인스턴스에서 EventPublishConfig 필드를 업데이트합니다.

curl -X PATCH \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  https://datafusion.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/instances/INSTANCE_ID?updateMask=event_publish_config \
  -d '{
    "event_publish_config": {
      "enabled": true,
      "topic": "projects/PROJECT_ID/topics/TOPIC_ID"
    }
}'

다음을 바꿉니다.

PROJECT_ID: Google Cloud 프로젝트 ID입니다.
LOCATION: 프로젝트의 위치
INSTANCE_ID: Cloud Data Fusion 인스턴스의 ID
TOPIC_ID: Pub/Sub 주제의 ID

인스턴스에서 이벤트 게시 삭제

인스턴스에서 이벤트 게시를 삭제하려면 이벤트 게시 enabled 값을 false로 업데이트합니다.

curl -X PATCH \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \ "https://datafusion.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/instances/INSTANCE_ID?updateMask=event_publish_config" \
  -d '{ "event_publish_config": { "enabled": false } }'

Pub/Sub 메시지를 읽는 함수 만들기

Cloud Run 함수는 Pub/Sub 메시지를 읽고 실패한 파이프라인 재시도와 같은 작업을 수행할 수 있습니다. Cloud Run Functions를 만들려면 다음 안내를 따르세요.

콘솔에서 Cloud Run Functions 페이지로 이동합니다. Google Cloud

Cloud Run 함수로 이동
함수 만들기를 클릭합니다.
함수 이름과 리전을 입력합니다.
트리거 유형 필드에서 Cloud Pub/Sub를 선택합니다.
Pub/Sub 주제 ID를 입력합니다.
다음을 클릭합니다.
Pub/Sub 메시지를 읽고 다른 작업을 수행하는 함수를 추가합니다. 예를 들어 다음 사용 사례에 함수를 추가할 수 있습니다.
- 파이프라인 오류에 대한 알림을 보냅니다.
- 레코드 수 또는 실행 정보와 같은 KPI에 대한 알림을 보냅니다.
- 다시 실행되지 않은 실패한 파이프라인을 다시 시작합니다.
Cloud Run Functions 예시는 사용 사례 섹션을 참조하세요.
배포를 클릭합니다. 자세한 내용은 Cloud Run 함수 배포를 참조하세요.

사용 사례: 파이프라인 상태 문서화 및 실패한 파이프라인 재시도

다음 Cloud Run 함수 예시는 파이프라인 실행 상태에 대한 Pub/Sub 메시지를 읽은 다음 Cloud Data Fusion에서 실패한 파이프라인을 재시도합니다.

이러한 예시 함수는 다음 Google Cloud 구성요소를 참조합니다.

Google Cloud 프로젝트: Cloud Run Functions 및 Pub/Sub 주제가 생성되는 프로젝트
Pub/Sub 주제: Cloud Data Fusion 인스턴스에 연결된 Pub/Sub 주제
Cloud Data Fusion 인스턴스: 파이프라인을 설계하고 실행하는 Cloud Data Fusion 인스턴스
BigQuery 테이블: 파이프라인 상태와 실행 및 재실행 세부정보를 캡처하는 BigQuery 테이블
Cloud Run Functions: 실패한 파이프라인을 재시도하는 코드를 배포하는 Cloud Run Functions

다음 Cloud Run Functions 예시는 Cloud Data Fusion 상태 이벤트에 대한 Pub/Sub 메시지를 읽습니다.

# Triggered from a message on a Pub/Sub topic.
@functions_framework.cloud_event
def cdf_event_trigger(cloud_event):

decoded_message = base64.b64decode(cloud_event.data["message"]["data"]).decode('utf-8') # Decode Pub/Sub message.
pubsub_message = json.loads(decoded_message)

# Extract pipeline run details.
projectName = pubsub_message["projectName"]
publishTime = pubsub_message["publishTime"]
instanceName = pubsub_message["instanceName"]
namespace = pubsub_message["programStatusEventDetails"]["namespace"]
applicationName = pubsub_message["programStatusEventDetails"]["applicationName"]
status = pubsub_message["programStatusEventDetails"]["status"]
event_timestamp = pd.to_datetime(pubsub_message["programStatusEventDetails"]["eventTime"], unit = 'ms')

print(f"projectName: {projectName}")
print(f"publishTime: {publishTime}")
print(f"instanceName: {instanceName}")
print(f"namespace: {namespace}")
print(f"applicationName: {applicationName}")
print(f"status: {status}")
print(f"event timestamp: {event_timestamp}")
try:
    error = pubsub_message["programStatusEventDetails"]["error"]
    print(f"error: {error}")
except:
    print(f"Pipeline: {applicationName}'s current status: {status}")

다음 예시 함수는 BigQuery 테이블을 만들고 저장하며 파이프라인 실행 세부정보를 쿼리합니다.

# Global variables.
pipeline_rerun_count = 0
has_pipeline_failed_and_rerun_recently = False # Timeframe: within last 60 minutes.
table_id = "bigquery-table-1" # The BigQuery target table for storing pipeline run information.

# Update BigQuery table with the pipeline status and rerun details.
schema=[
    bigquery.SchemaField("Project_Name", "STRING"),
    bigquery.SchemaField("Instance_Name", "STRING"),
    bigquery.SchemaField("Namespace", "STRING"),
    bigquery.SchemaField("Pipeline_Name", "STRING"),
    bigquery.SchemaField("Pipeline_Status", "STRING"),
    bigquery.SchemaField("Event_Timestamp", "TIMESTAMP"),
    bigquery.SchemaField("Pipeline_Rerun_Count", "INTEGER"),
]

# Prepare DataFrame to load the data in BigQuery.
data = {'Project_Name':[projectName], 'Instance_Name':[instanceName], 'Namespace':[namespace], 'Pipeline_Name':[applicationName], 'Pipeline_Status':[status], 'Event_Timestamp':[event_timestamp], 'Pipeline_Rerun_Count':[pipeline_rerun_count]}
dataframe = pd.DataFrame(data)

# Prepare BigQuery data load job configuration.
job_config = bigquery.LoadJobConfig(schema=schema)

job = bq_client.load_table_from_dataframe(dataframe, table_id, job_config=job_config)
job.result()  # Wait for the job to complete.

table = bq_client.get_table(table_id)  # Make an API request.
print("BigQuery table: {} updated.".format(table_id))

다음 예시 함수는 실패한 파이프라인과 지난 1시간 동안 다시 실행되었는지 확인합니다.

bq_client = bigquery.Client()

if status == "FAILED":
    print(f"ALERT -- Pipeline: {applicationName} has failed. Checking for rerun: pipeline hasn't failed and rerun in the last 60 minutes.")

    QUERY = f"""
        SELECT * FROM `{table_id}`
        WHERE Pipeline_Name = "{applicationName}" AND Pipeline_Status = "FAILED"
        AND "{event_timestamp}" < DATETIME_ADD(Event_Timestamp, INTERVAL 60 MINUTE)
        AND Pipeline_Rerun_Count > 0
        """

    query_job = bq_client.query_and_wait(QUERY)  # API request.
    row_count = query_job.total_rows  # Waits for query to finish.
    print(f"Query job result row count: {row_count}")

    if (row_count > 0):
        print("Pipeline has FAILED and rerun recently...")
        global has_pipeline_failed_and_rerun_recently
        has_pipeline_failed_and_rerun_recently = True

실패한 파이프라인이 최근에 실행되지 않은 경우 다음 예시 함수는 실패한 파이프라인을 다시 실행합니다.

if not has_pipeline_failed_and_rerun_recently:
    applicationName = applicationName
    auth_token = get_access_token()
    post_headers = {"Authorization": "Bearer " + auth_token,"Accept": "application/json"}
    cdap_endpoint = "https://instance1-project1-dot-location1.datafusion.googleusercontent.com/api"
    run_pipeline_endpoint = cdap_endpoint + "/v3/namespaces/{}/apps/{}/workflows/DataPipelineWorkflow/start".format(namespace, applicationName)

    # Start the job.
    response = requests.post(run_pipeline_endpoint,headers=post_headers)
    print(f"Response for restarting the failed pipeline: {response}")
    global pipeline_rerun_count
    pipeline_rerun_count = 1

다음 단계

Cloud Run Functions를 작성하는 방법을 알아봅니다.