apache-beam

0열

1답변

Apache Beam Dataflow : 'NoneType'객체에 'parts'속성이 없습니다.

저는 파이프 라인을 작성하여 pubsub에서 스트림을 읽고 apache 빔을 사용하여 Google Cloud Dataflow를 사용하여 bigquery에 씁니다. 이 코드가 있습니다 import apache_beam as beam from apache_beam.transforms.window import FixedWindows topic = 'proj

1열

1답변

DoFn의 HTTP 클라이언트

DataFlow에서 실행되는 Apache Beam Pipeline에 대해 DoFn을 통해 POST를 요청하고 싶습니다. 이를 위해 PoolingHttpClientConnectionManager에 구성된 HttpClosableClient를 인스턴스화하는 클라이언트를 만들었습니다. 그러나 내가 처리하는 각 요소에 대해 클라이언트를 인스턴스화합니다. 내 모든 요

1열

1답변

테이블 행 PCollection을 키, 값 PCollections로 변환하는 방법은 무엇입니까?

유형 II를 들어, .CoGroupByKey() 상황 기본적으로 나는 두 개의 큰 pCollections을 가지고 있고이 둘 사이의 차이를 찾을 수 있어야에 입력에 필요한 pCollections에 pCollections을 변환하는 방법에 대한 설명서가 없습니다 ETL이 변경되면 (pColl1에없는 경우 pColl2에서 중첩 된 필드에 추가) BigQuery

1열

1답변

FileBasedSink의 유형 매개 변수는 무엇입니까?

FileBasedSink를 버전 2.0.0에서 2.2.0으로 확장하는 사용자 지정 싱크를 마이그레이션하려고합니다. 이 클래스는 변경 및 두 개의 여분의 형태 파라미터를 추가했습니다 : UserT 및 DestinationT : 나는 FileBasedSink의 문서를 확인했지만 그것의 목적을 찾을 수 없습니다 @Experimental(value=FILESYST

0열

1답변

아파치 빔 내가 차원 테이블을로드 할 ETL을 구축을 위해 노력하고 있어요 PCollection

에 일련 번호를 추가합니다. 저는 Apache Bea와 Python, DataFlow 및 BigQuery를 사용하고 있습니다. 나는 BigQuery에로의로드하기 위해 pcollection의 각 요소에 일련 번호를 지정해야하지만이 작업을 수행 할 수있는 방법을 찾을 수 없습니다. DataFlow가 이전 집계를 만들기 위해 필요하다고 생각하고 시퀀스 번호를 추

0열

1답변

이 같은 약 보이는 파이프 라인이 데이터 흐름

구글을 통해 아파치 빔 파이썬을 사용하여 여러 노동자들에게 작은 수집 출력을 배포 현재 분할을 지원합니다. some_expensive_fn은 실행하는 데 시간이 걸릴 수있는 값 비싼 작업입니다. some_inexpensive_agg은 아마도이 질문에 중요하지 않지만 집계 용도로 몇 가지 결과가 있음을 보여주는 것입니다. SomeSourceProducingL

0열

1답변

Google 데이터 흐름 템플릿 - NumWorkers, MaxNumWorkers, WorkerMachineType, AutoscalingAlgorithm 전달하기

Google Cloud Dataflow Java SDK 2.1.0을 사용하고 있습니다. 런타임에 전달하려는 다른 ValueProvider 파이프 라인 옵션과 함께 --templateLocation 파이프 라인 옵션을 사용하여 템플릿을 만들 수 있습니다. 클라우드 기능 내에서 Google dataflow API를 사용하여 템플릿을 호출하고 있습니다. 템플릿은

0열

1답변

java.lang.IllegalStateException : 데이터 흐름 2.X의 기본 코더를 반환 할 수 없습니다.

데이터 흐름 2.1 SDK에 간단한 파이프 라인이 있습니다. pubsub에서 데이터를 읽은 다음 DoFn을 적용합니다. PCollection<MyClass> e = streamData.apply("ToE", ParDo.of(new MyDoFNClass())); 이 파이프 라인에 오류가 아래 방법 : java.lang.IllegalStateExceptio

1열

1답변

아파치 빔 : DoFn과 PTransform

DoFn과 PTransform은 모두 PCollection에 대한 작업을 정의하는 수단입니다. 우리는 언제 어느 것을 사용해야하는지 어떻게 알 수 있습니까?

0열

1답변

아파치 빔 2.2 pipeline.apply 이러한 방법의 예외

public static void main(String[] args) { //Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create()); DataflowPipelineOptions options = PipelineOption