Python을 사용하여 Dataflow 파이프라인 만들기

이 빠른 시작에서는 Python용 Apache Beam SDK를 사용하여 파이프라인을 정의하는 프로그램을 빌드하는 방법을 알아봅니다. 그런 다음 직접 로컬 실행기 또는 Dataflow와 같은 클라우드 기반 실행기를 사용하여 파이프라인을 실행합니다. WordCount 파이프라인 소개는 Apache Beam에서 WordCount를 사용하는 방법 동영상을 참조하세요.

Google Cloud 콘솔에서 이 태스크에 대한 단계별 안내를 직접 수행하려면 둘러보기를 클릭합니다.

둘러보기

시작하기 전에

  1. Google Cloud 계정에 로그인합니다. Google Cloud를 처음 사용하는 경우 계정을 만들고 Google 제품의 실제 성능을 평가해 보세요. 신규 고객에게는 워크로드를 실행, 테스트, 배포하는 데 사용할 수 있는 $300의 무료 크레딧이 제공됩니다.
  2. Install the Google Cloud CLI.

  3. To initialize the gcloud CLI, run the following command: 

    gcloud init

  4. Create or select a Google Cloud project.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

  5. Google Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다.

  6. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Datastore, and Cloud Resource Manager APIs: 

    gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com cloudresourcemanager.googleapis.com

  7. Create local authentication credentials for your user account: 

    gcloud auth application-default login

  8. Google 계정에 역할을 부여합니다. 다음 각 IAM 역할에 대해 다음 명령어를 한 번씩 실행합니다. roles/iam.serviceAccountUser 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE
    • PROJECT_ID를 프로젝트 ID로 바꿉니다.
    • EMAIL_ADDRESS를 이메일 주소로 바꿉니다.
    • ROLE을 각 개별 역할로 바꿉니다.
    9.
  9. Install the Google Cloud CLI.

  10. To initialize the gcloud CLI, run the following command: 

    gcloud init

  11. Create or select a Google Cloud project.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

  12. Google Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다.

  13. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Datastore, and Cloud Resource Manager APIs: 

    gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com cloudresourcemanager.googleapis.com

  14. Create local authentication credentials for your user account: 

    gcloud auth application-default login

  15. Google 계정에 역할을 부여합니다. 다음 각 IAM 역할에 대해 다음 명령어를 한 번씩 실행합니다. roles/iam.serviceAccountUser 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE
    • PROJECT_ID를 프로젝트 ID로 바꿉니다.
    • EMAIL_ADDRESS를 이메일 주소로 바꿉니다.
    • ROLE을 각 개별 역할로 바꿉니다.
    16.

  16. Compute Engine 기본 서비스 계정에 역할을 부여합니다. 다음 IAM 역할마다 다음 명령어를 1회 실행합니다.

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.objectAdmin
    gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" --role=SERVICE_ACCOUNT_ROLE
    • PROJECT_ID를 프로젝트 ID로 바꿉니다.
    • 여기에서 PROJECT_NUMBER를 프로젝트 번호로 바꿉니다. 프로젝트 번호를 찾으려면 프로젝트 식별을 참조하거나 gcloud projects describe 명령어를 사용합니다.
    • SERVICE_ACCOUNT_ROLE을 각 개별 역할로 바꿉니다.
  17. Cloud Storage 버킷을 만들고 다음과 같이 구성합니다.
    • 스토리지 클래스를 S(스탠더드)로 설정합니다.
    • 스토리지 위치를 US(미국)으로 설정합니다.
    • BUCKET_NAME을 고유한 버킷 이름으로 바꿉니다. 버킷 네임스페이스는 전역적이며 공개로 표시되므로 버킷 이름에 민감한 정보를 포함해서는 안 됩니다. 
    gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US
  18. Google Cloud 프로젝트 ID와 Cloud Storage 버킷 이름을 복사합니다. 이 값은 이 문서의 뒷부분에서 필요합니다.

환경 설정

이 섹션에서는 명령 프롬프트를 사용하여 venv를 사용해 파이프라인 프로젝트를 실행할 격리된 Python 가상 환경을 설정합니다. 이 프로세스를 통해 한 프로젝트의 종속 항목을 다른 프로젝트의 종속 항목에서 분리할 수 있습니다.

명령 프롬프트를 사용할 수 없으면 Cloud Shell을 사용해도 됩니다. Cloud Shell에는 이미 Python 3용 패키지 관리자가 설치되어 있으므로 가상 환경 만들기로 건너뛸 수 있습니다.

Python을 설치한 후 가상 환경을 만들려면 다음 단계별 안내를 따르세요.

  1. Python 3 및 pip가 시스템에서 실행 중인지 확인합니다.
    python --version
python -m pip --version
  2. 필요한 경우 Python 3을 설치한 후 Python 가상 환경을 설정합니다. Python 설치 및 venv 설정 섹션의 Python 개발 환경 설정 페이지를 참조하세요. Python 3.10 이상을 사용하는 경우 Dataflow Runner v2를 사용 설정해야 합니다. Runner v1을 사용하려면 Python 3.9 이하를 사용합니다.

빠른 시작을 완료한 후에는 deactivate를 실행하여 가상 환경을 중지할 수 있습니다.

Apache Beam SDK 가져오기

Apache Beam SDK는 데이터 파이프라인용 오픈소스 프로그래밍 모델입니다. Apache Beam 프로그램으로 파이프라인을 정의한 다음 Dataflow와 같은 실행기를 선택하여 파이프라인을 실행하게 됩니다.

Apache Beam SDK를 다운로드하고 설치하려면 다음 단계별 안내를 따르세요.

  1. 이전 섹션에서 만든 Python 가상 환경에 위치하고 있는지 확인하세요. 메시지가 <env_name>으로 시작되는지 확인합니다. 여기서 env_name은 가상 환경의 이름입니다.
  2. Python wheel 패키징 표준을 설치합니다.
    pip install wheel
  3. 최신 버전의 Python용 Apache Beam SDK를 설치합니다.
    4. pip install 'apache-beam[gcp]'

    Microsoft Windows에서는 다음 명령어를 사용합니다.

    pip install apache-beam[gcp]

    연결 상태에 따라 설치에 다소 시간이 걸릴 수 있습니다.

로컬에서 파이프라인 실행

파이프라인이 로컬에서 실행되는 방법을 보려면 apache_beam 패키지에 포함된 wordcount 예시의 바로 활용할 수 있는 Python 모듈을 사용합니다.

wordcount 파이프라인 예시는 다음을 수행합니다.

  1. 텍스트 파일을 입력으로 가져옵니다.

    이 텍스트 파일은 리소스 이름 gs://dataflow-samples/shakespeare/kinglear.txt으로 Cloud Storage 버킷에 저장됩니다.

  2. 각 줄을 단어로 파싱합니다.
  3. 토큰화된 단어에서 빈도 카운트를 수행합니다.

wordcount 파이프라인을 로컬에서 스테이징하려면 다음 단계별 안내를 따르세요.

  1. 로컬 터미널에서 wordcount 예시를 실행합니다. 
    python -m apache_beam.examples.wordcount \
  --output outputs
  2. 파이프라인의 출력을 확인합니다. 
    more outputs*
  3. 종료하려면 q를 누릅니다.
파이프라인을 로컬에서 실행하면 Apache Beam 프로그램을 테스트하고 디버깅할 수 있습니다. Apache Beam GitHub에서 wordcount.py 소스 코드를 확인할 수 있습니다.

Dataflow 서비스에서 파이프라인 실행

이 섹션에서는 Dataflow 서비스의 apache_beam 패키지에서 wordcount 예시 파이프라인을 실행합니다. 이 예시에서는 DataflowRunner를 --runner의 매개변수로 지정합니다.
  • 파이프라인 실행하기 
    python -m apache_beam.examples.wordcount \
    --region DATAFLOW_REGION \
    --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --temp_location gs://BUCKET_NAME/tmp/

    다음을 바꿉니다.

    • DATAFLOW_REGION: Dataflow 작업을 배포할 리전입니다(예: europe-west1).

      --region 플래그는 메타데이터 서버, 로컬 클라이언트 또는 환경 변수에 설정된 기본 리전을 재정의합니다.

    • BUCKET_NAME: 이전에 복사한 Cloud Storage 버킷 이름입니다.
    • PROJECT_ID: 이전에 복사한 Google Cloud 프로젝트 ID입니다.

결과 보기

Dataflow를 사용하여 파이프라인을 실행한 경우 결과는 Cloud Storage 버킷에 저장됩니다. 이 섹션에서는 Google Cloud 콘솔 또는 로컬 터미널을 사용하여 파이프라인이 실행 중인지 확인합니다.

Google Cloud 콘솔

Google Cloud 콘솔에서 결과를 보려면 다음 단계를 따르세요.

  1. Google Cloud 콘솔에서 Dataflow 작업 페이지로 이동합니다.

    작업으로 이동

    작업 페이지에는 wordcount 작업의 상세 정보가 표시되는데, 처음에는 실행 중이 나타나고 다음에는 성공으로 상태가 바뀝니다.

  2. Cloud Storage 버킷 페이지로 이동합니다.

    버킷으로 이동

  3. 프로젝트의 버킷 목록에서 앞서 만든 스토리지 버킷을 클릭합니다.

    wordcount 디렉터리에 작업을 통해 만든 출력 파일이 표시됩니다.

로컬 터미널

터미널 또는 Cloud Shell을 사용하여 결과를 확인합니다.

  1. 출력 파일을 나열하려면 gcloud storage ls 명령어를 사용합니다.
    gcloud storage ls gs://BUCKET_NAME/results/outputs* --long

    2. BUCKET_NAME을 파이프라인 프로그램에 사용되는 Cloud Storage 버킷 이름으로 바꿉니다.

  2. 출력 파일의 결과를 보려면 gcloud storage cat 명령어를 사용합니다. 
    gcloud storage cat gs://BUCKET_NAME/results/outputs*

파이프라인 코드 수정

이전 예시의 wordcount 파이프라인은 대소문자를 구분합니다. 다음 단계에서는 wordcount 파이프라인이 대소문자를 구분하지 않도록 파이프라인을 수정하는 방법을 보여줍니다.
  1. 로컬 머신의 Apache Beam GitHub 저장소에서 wordcount 코드의 최신 사본을 다운로드합니다.
  2. 로컬 터미널에서 파이프라인을 실행합니다. 
    python wordcount.py --output outputs
  3. 결과 보기 
    more outputs*
  4. 종료하려면 q를 누릅니다.
  5. 원하는 편집기에서 wordcount.py 파일을 엽니다.
  6. run 함수에서 파이프라인 단계를 검사합니다. 
    counts = (
        lines
        | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
        | 'GroupAndSum' >> beam.CombinePerKey(sum))

    split하면 줄이 문자열 단어로 분리됩니다.

  7. 문자열을 소문자로 입력하려면 split 다음의 줄을 수정하세요. 
    counts = (
        lines
        | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
        | 'lowercase' >> beam.Map(str.lower)
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
        | 'GroupAndSum' >> beam.CombinePerKey(sum))
    이렇게 수정하면 str.lower 함수가 모든 단어에 매핑됩니다. 이 줄은 beam.Map(lambda word: str.lower(word))와 동일합니다.
  8. 파일을 저장하고 수정된 wordcount 작업을 실행합니다. 
    python wordcount.py --output outputs
  9. 수정된 파이프라인의 결과를 확인합니다. 
    more outputs*
  10. 종료하려면 q를 누릅니다.
  11. Dataflow 서비스에서 수정된 파이프라인을 실행합니다. 
    python wordcount.py \
    --region DATAFLOW_REGION \
    --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --temp_location gs://BUCKET_NAME/tmp/

    다음을 바꿉니다.

    • DATAFLOW_REGION: Dataflow 작업을 배포할 리전입니다.
    • BUCKET_NAME: Cloud Storage 버킷 이름입니다.
    • PROJECT_ID: Google Cloud 프로젝트 ID입니다.

삭제

이 페이지에서 사용한 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 리소스가 포함된 Google Cloud 프로젝트를 삭제하면 됩니다.

  1. In the Google Cloud console, go to the Cloud Storage Buckets page.

    Go to Buckets

  2. Click the checkbox for the bucket that you want to delete.
  3. To delete the bucket, click Delete, and then follow the instructions.

  4. 프로젝트를 유지할 경우 Compute Engine 기본 서비스 계정에 부여한 역할을 취소합니다. 다음 IAM 역할마다 다음 명령어를 1회 실행합니다.

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.objectAdmin
    gcloud projects remove-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=SERVICE_ACCOUNT_ROLE

  5. Optional: Revoke the authentication credentials that you created, and delete the local credential file. 

    gcloud auth application-default revoke

  6. Optional: Revoke credentials from the gcloud CLI. 

    gcloud auth revoke

다음 단계