apache-beam-io

0열

1답변

나는 현재 Google tutorial 같은 아파치 빔 SDK의 V2.1.0를 기반으로 데이터 흐름 템플릿을 만들려고 해요 이는 경우 내 메인 클래스 public static void main(String[] args) { // Initialize options DispatcherOptions options = PipelineOption

1열

1답변

DoFn의 HTTP 클라이언트

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

0열

1답변

데이터 흐름 GroupBy -> 키 기반 다중 출력

그룹 키를 기반으로 GroupBy의 출력을 여러 출력 파일로 리디렉션 할 수있는 간단한 방법이 있습니까? Bin.apply(GroupByKey.<String, KV<Long,Iterable<TableRow>>>create()) .apply(ParDo.named("Print Bins").of(...) .apply(TextIO.Write.to(*Output

0열

1답변

Apache Beam count HBase 행 블록 및 반환하지 않음

Apache Beam을시 도하 고 HBase 테이블을 읽고 계산하는 데 사용해보십시오. Count.globally없이 테이블을 읽으려고하면 행을 읽을 수 있지만 행 수가 세어지면 프로세스가 중단되고 절대로 종료하지 않습니다. Pipeline p = Pipeline.create(options); p.apply("read", HBaseIO.read().wi

1열

2답변

Apache Beam을 사용하여 Hive에서 읽는 방법은 무엇입니까?

Apache Beam을 사용하여 Hive를 읽는 방법/Apache Beam에서 Hive를 소스로 사용하는 방법은 무엇입니까?

0열

1답변

아파치 빔 KafkaIO 외부 데이터 저장소에 대한 오프셋 관리

아파치 빔에 KafkaIO를 사용하여 여러 카프카 브로커에서 읽으려고합니다. 오프셋 관리의 기본 옵션은 kafka 파티션 자체입니다 (더 이상 kafka> 0.9에서 동물원을 사용하지 않음). 이 설정으로 작업/파이프 라인을 다시 시작할 때 중복 된 레코드 누락과 관련된 문제가 있습니다. 내가 읽은 바로는이를 처리하는 가장 좋은 방법은 외부 데이터 저장소에

1열

2답변

Apache Beam을 사용하여 데이터베이스에서 대량 데이터 읽기

제 쿼리가 수백만 행을 반환하는 경우 JdbcIO가 병렬로 쿼리를 실행하는 방법을 알고 싶습니다. https://issues.apache.org/jira/browse/BEAM-2803과 관련된 풀 요청을 언급했습니다. 나는 그것을 완전히 이해할 수 없었다. ReadAllexpand 메서드는 ParDo을 사용합니다. 따라서 병렬로 데이터를 읽으려면 데이터베이

0열

1답변

Maven을 사용하지 않는 Apache Beam 프로그램 실행

Apache Spark 러너를 사용하는 간단한 예제 빔 프로그램을 실행하고 싶습니다. 1) 로컬에서 프로그램을 성공적으로 컴파일 할 수있었습니다. 2) Maven이 설치되어 있지 않은 QA 상자에 JAR 파일을 푸시합니다. 3) 예제 프로그램을 컴파일하고 실행하는 Maven 명령어로 예제를 볼 수 있습니다. 4) Maven을 설치하지 않고 코드를 실행하는

0열

1답변

모듈 객체에는 속성이 없습니다. BigqueryV2 - 로컬 Apache Beam

빔 제공 I/O API를 사용하여 Apache Beam과 함께 로컬에서 파이프 라인을 실행하려고합니다. 나는 가상 환경을 사용하여 내 환경을 Beam Python quickstart으로 제안했고 wordcount.py 예제를 실행할 수 있습니다. beam.Create 및 beam.ParDo으로 사용자 지정 파이프 라인을 제대로 실행할 수도 있습니다. 그러나

1열

1답변

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

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