Dataflow 템플릿을 사용하여 스트리밍 파이프라인 만들기

이 빠른 시작에서는 Google에서 제공하는 Dataflow 템플릿을 사용하여 스트리밍 파이프라인을 만드는 방법을 설명합니다. 특히 이 빠른 시작에서는 Pub/Sub Topic to BigQuery 템플릿을 예시로 사용합니다.

Pub/Sub Topic to BigQuery 템플릿은 Pub/Sub 주제에서 JSON 형식의 메시지를 읽고 BigQuery 테이블에 쓰는 스트리밍 파이프라인입니다.

Google Cloud Console에서 이 작업에 대한 단계별 안내를 직접 수행하려면 둘러보기를 클릭합니다.

둘러보기

시작하기 전에

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

  2. Google Cloud Console의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.

    프로젝트 선택기로 이동

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

  4. API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Pub/Sub, and Resource Manager 사용 설정

    API 사용 설정

    5.

  5. Google Cloud Console의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.

    프로젝트 선택기로 이동

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

  7. API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, BigQuery, Pub/Sub, and Resource Manager 사용 설정

    API 사용 설정

    8.
  8. Cloud Storage 버킷을 만듭니다.
    1. Google Cloud Console에서 Cloud Storage 버킷 페이지로 이동합니다.

      버킷 페이지로 이동

    2. 버킷 만들기를 클릭합니다.
    3. 버킷 만들기 페이지에서 버킷 정보를 입력합니다. 다음 단계로 이동하려면 계속을 클릭합니다.
      • 버킷 이름 지정에서 고유한 버킷 이름을 입력합니다. 버킷 네임스페이스는 전역적이며 공개로 표시되므로 버킷 이름에 민감한 정보를 포함해서는 안 됩니다.
      • 데이터를 저장할 위치 선택에서 다음을 수행합니다.
        • 위치 유형 옵션을 선택합니다.
        • 위치 옵션을 선택합니다.
      • 데이터의 기본 스토리지 클래스 선택에서 다음을 선택합니다. 표준.
      • 객체 액세스를 제어하는 방식 선택에서 액세스 제어 옵션을 선택합니다.
      • 고급 설정(선택사항)에서 암호화 방법, 보관 정책 또는 버킷 라벨을 지정합니다.
    4. 만들기를 클릭합니다.
  9. 뒤에 나오는 섹션에서 필요하므로 다음을 복사합니다.
    • Cloud Storage 버킷 이름
    • Google Cloud 프로젝트 ID

      이 ID를 찾으려면 프로젝트 식별을 참조하세요.

  10. 빠른 시작 단계를 완료하려면 사용자 계정에 Dataflow 관리자 역할 및 서비스 계정 사용자 역할이 있어야 합니다. Compute Engine 기본 서비스 계정에는 Dataflow 작업자 역할이 있어야 합니다. Google Cloud 콘솔에서 필요한 역할을 추가하려면 다음 안내를 따르세요.

    1. Google Cloud 콘솔에서 IAM 페이지로 이동합니다.
      IAM으로 이
    2. 프로젝트를 선택합니다.
    3. 사용자 계정이 포함된 행에서 주 구성원 수정을 클릭한 다음 다른 역할 추가를 클릭합니다.
    4. 드롭다운 목록에서 Dataflow 관리자 역할을 선택합니다.
    5. 서비스 계정 사용자 역할에도 같은 단계를 반복한 후 저장을 클릭합니다.
    6. Compute Engine 기본 서비스 계정이 포함된 행에서 주 구성원 수정을 클릭한 다음 다른 역할 추가를 클릭합니다.
    7. 드롭다운 목록에서 Dataflow 작업자 역할을 선택합니다.
    8. Pub/Sub 편집자 및 BigQuery 데이터 편집자 역할에도 같은 단계를 반복한 후 저장을 클릭합니다.

      역할 부여에 대한 상세 설명은 콘솔을 사용하여 IAM 역할 부여를 참조하세요.

  11. 기본적으로 각각의 새 프로젝트는 기본 네트워크로 시작합니다. 프로젝트에 대한 기본 네트워크가 사용 중지되었거나 삭제된 경우 프로젝트에 사용자 계정에 Compute Network 사용자 역할(roles/compute.networkUser)이 있는 네트워크가 있어야 합니다.

BigQuery 데이터세트와 테이블 만들기

Google Cloud 콘솔을 사용하여 Pub/Sub 주제에 적절한 스키마로 BigQuery 데이터 세트와 테이블을 만듭니다.

이 예시에서 데이터 세트 이름은 taxirides이고 테이블 이름은 realtime입니다. 이 데이터 세트와 테이블을 만들려면 다음 단계를 따르세요.

  1. Google Cloud Console에서 BigQuery 페이지로 이동합니다.
    BigQuery로 이동
  2. 데이터 세트를 만들 프로젝트 옆에 있는 탐색기 패널에서 작업 보기를 클릭한 후 데이터 세트 만들기를 클릭합니다.
  3. 데이터 세트 만들기 패널에서 다음 단계를 수행합니다.
    1. 데이터 세트 ID에 taxirides를 입력합니다. 데이터 세트 ID는 Google Cloud 프로젝트마다 고유합니다.
    2. 데이터 위치로 미국(US)을 선택합니다. 공개 데이터 세트는 US 멀티 리전 위치에 저장됩니다. 편의상 같은 위치에 데이터 세트를 배치합니다.
    3. 다른 기본 설정은 그대로 두고 데이터 세트 만들기를 클릭합니다.
  4. 탐색기 패널에서 프로젝트를 펼칩니다.
  5. taxirides 데이터 세트 옆의 작업 보기를 클릭한 후 테이블 만들기를 클릭합니다.
  6. 테이블 만들기 패널에서 다음 단계를 수행합니다.
    1. 소스 섹션에서 다음 항목으로 테이블 만들기에 빈 테이블을 선택합니다.
    2. 대상 섹션의 테이블에 realtime을 입력합니다.
    3. 스키마 섹션에서 텍스트로 수정 전환 버튼을 클릭하고 다음 스키마 정의를 상자에 붙여넣습니다. 
      ride_id:string,point_idx:integer,latitude:float,longitude:float,timestamp:timestamp,
meter_reading:float,meter_increment:float,ride_status:string,passenger_count:integer
    4. 파티션 나누기 및 클러스터 설정 섹션에서 파티션 나누기에 timestamp 필드를 선택합니다.
  7. 다른 기본 설정은 그대로 두고 테이블 만들기를 클릭합니다.

파이프라인 실행하기

Google에서 제공하는 Pub/Sub Topic to BigQuery 템플릿을 사용하여 스트리밍 파이프라인을 실행합니다. 파이프라인은 입력 주제에서 수신되는 데이터를 가져옵니다.

  1. Google Cloud 콘솔에서 Dataflow 작업 페이지로 이동합니다.
    작업으로 이동
  2. 템플릿에서 작업 만들기를 클릭합니다.
  3. taxi-data를 Dataflow 작업의 작업 이름으로 입력합니다.
  4. Dataflow 템플릿에 BigQuery에 대한 Pub/Sub 주제 템플릿을 선택합니다.
  5. 입력 Pub/Sub 주제에서 직접 주제 입력을 클릭합니다.
  6. 대화상자의 주제 이름에 다음을 입력하고 저장을 클릭합니다. 
    projects/pubsub-public-data/topics/taxirides-realtime

    공개적으로 사용 가능한 Pub/Sub 주제는 뉴욕시 택시 및 리무진 조합의 공개 데이터 세트를 기반으로 합니다. 다음은 JSON 형식으로 된 이 주제의 샘플 메시지입니다.

    {
  "ride_id": "19c41fc4-e362-4be5-9d06-435a7dc9ba8e",
  "point_idx": 217,
  "latitude": 40.75399,
  "longitude": -73.96302,
  "timestamp": "2021-03-08T02:29:09.66644-05:00",
  "meter_reading": 6.293821,
  "meter_increment": 0.029003782,
  "ride_status": "enroute",
  "passenger_count": 1
}
  7. BigQuery 출력 테이블에 다음을 입력합니다. 
    PROJECT_ID:taxirides.realtime

    PROJECT_ID를 BigQuery 데이터 세트를 만든 프로젝트의 프로젝트 ID로 바꿉니다.

  8. 임시 위치에 다음을 입력합니다. 
    BUCKET_NAME/temp/

    BUCKET_NAME을 Cloud Storage 버킷 이름으로 바꿉니다. temp 폴더는 스테이징된 파이프라인 작업과 같은 임시 파일을 저장합니다.

  9. 프로젝트에 기본 네트워크가 없는 경우, 선택적 매개변수 표시를 클릭한 후 네트워크 및 서브네트워크를 입력합니다. 자세한 내용은 네트워크 및 서브네트워크 지정을 참조하세요.
  10. 작업 실행을 클릭합니다.

결과 보기

realtime 테이블에 기록된 데이터를 보려면 다음 단계를 따르세요.

  1. Google Cloud Console에서 BigQuery 페이지로 이동합니다.

    BigQuery로 이동

  2. 쿼리 편집기에 다음 쿼리를 붙여넣습니다.

    SELECT * FROM `PROJECT_ID.taxirides.realtime`
WHERE `timestamp` > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
LIMIT 1000

    PROJECT_ID를 BigQuery 데이터 세트를 만든 프로젝트의 프로젝트 ID로 바꿉니다. 데이터가 테이블에 표시되는 데 최대 1분이 걸릴 수 있습니다.

  3. 실행을 클릭합니다.

    이 쿼리에서는 지난 24시간 동안 테이블에 추가된 행을 반환합니다. 표준 SQL을 사용하여 쿼리를 실행할 수도 있습니다.

삭제

이 페이지에서 사용한 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 다음 단계를 수행합니다.

프로젝트 삭제

비용이 청구되지 않도록 하는 가장 쉬운 방법은 빠른 시작에서 만든 Google Cloud 프로젝트를 삭제하는 것입니다.

  1. Google Cloud 콘솔에서 리소스 관리 페이지로 이동합니다.

    리소스 관리로 이동

  2. 프로젝트 목록에서 삭제할 프로젝트를 선택하고 삭제를 클릭합니다.
  3. 대화상자에서 프로젝트 ID를 입력한 후 종료를 클릭하여 프로젝트를 삭제합니다.

개별 리소스 삭제

이 빠른 시작에서 사용한 Google Cloud 프로젝트를 유지하려면 개별 리소스를 삭제합니다.

  1. Google Cloud 콘솔에서 Dataflow 작업 페이지로 이동합니다.
    작업으로 이동
  2. 작업 목록에서 스트리밍 작업을 선택합니다.
  3. 탐색에서 중지를 클릭합니다.
  4. 작업 중지 대화상자에서 파이프라인을 취소하거나 드레이닝한 다음 작업 중지를 클릭합니다.
  5. Google Cloud Console에서 BigQuery 페이지로 이동합니다.
    BigQuery로 이동
  6. 탐색기 패널에서 프로젝트를 펼칩니다.
  7. 삭제하려는 데이터 세트 옆에 있는 작업 보기를 클릭한 다음 열기를 클릭합니다.
  8. 세부정보 패널에서 데이터 세트 삭제를 클릭하고 안내를 따릅니다.
  9. Google Cloud 콘솔에서 Cloud Storage 버킷 페이지로 이동합니다.

    버킷으로 이동

  10. 삭제할 버킷의 체크박스를 클릭합니다.
  11. 버킷을 삭제하려면 삭제를 클릭한 후 안내를 따르세요.

다음 단계