4
Apache Beam과 DirectRunner을 러너로 사용하여 파이프 라인을 구축하고 있습니다. Google 클라우드 팝/하위에서Apache Beam 및 CombineFn의 코더 문제
- 풀 데이터의 고정 창을 사용하여 자바 객체로
- 윈도우 이벤트를
- 역 직렬화를 (현재 로컬로 실행하는 에뮬레이터를 사용하여) : 우리는 현재 우리가 이에 간단한 파이프 라인을 시도하고있다 1 분
- 이벤트를 이벤트 목록으로 변환하는 사용자 정의
CombineFn을 사용하여이 창을 결합하십시오.
파이프 번호 :
pipeline
.apply(PubsubIO.<String>read().topic(options.getTopic()).withCoder(StringUtf8Coder.of()))
.apply("ParseEvent", ParDo.of(new ParseEventFn()))
.apply("WindowOneMinute",Window.<Event>into(FixedWindows.of(Duration.standardMinutes(1))))
.apply("CombineEvents", Combine.globally(new CombineEventsFn()));
ParseEvent 기능 :
static class ParseEventFn extends DoFn<String, Event> {
@ProcessElement
public void processElement(ProcessContext c) {
String json = c.element();
c.output(gson.fromJson(json, Event.class));
}
}
CombineEvents 기능 : 거대한 코드에 대한
java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:293)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalStateException: Unable to return a default Coder for CombineEvents/Combine.perKey(CombineEvents)/Combine.GroupedValues/ParDo(Anonymous).out [PCollection]. Correct one of the following root causes:
No Coder has been manually specified; you may do so using .setCoder().
Inferring a Coder from the CoderRegistry failed: Unable to provide a default Coder for org.apache.beam.sdk.values.KV<K, OutputT>. Correct one of the following root causes:
Building a Coder using a registered CoderFactory failed: Cannot provide coder for parameterized type org.apache.beam.sdk.values.KV<K, OutputT>: Unable to provide a default Coder for java.lang.Object. Correct one of the following root causes:
Building a Coder using a registered CoderFactory failed: Cannot provide coder based on value with class java.lang.Object: No CoderFactory has been registered for the class.
Building a Coder from the @DefaultCoder annotation failed: Class java.lang.Object does not have a @DefaultCoder annotation.
Building a Coder from the fallback CoderProvider failed: Cannot provide coder for type java.lang.Object: [email protected] could not provide a Coder for type java.lang.Object: Cannot provide ProtoCoder because java.lang.Object is not a subclass of com.google.protobuf.Message; [email protected] could not provide a Coder for type java.lang.Object: Cannot provide SerializableCoder because java.lang.Object does not implement Serializable.
Building a Coder from the @DefaultCoder annotation failed: Class org.apache.beam.sdk.values.KV does not have a @DefaultCoder annotation.
Using the default output Coder from the producing PTransform failed: Unable to provide a default Coder for org.apache.beam.sdk.values.KV<K, OutputT>. Correct one of the following root causes:
Building a Coder using a registered CoderFactory failed: Cannot provide coder for parameterized type org.apache.beam.sdk.values.KV<K, OutputT>: Unable to provide a default Coder for java.lang.Object. Correct one of the following root causes:
Building a Coder using a registered CoderFactory failed: Cannot provide coder based on value with class java.lang.Object: No CoderFactory has been registered for the class.
Building a Coder from the @DefaultCoder annotation failed: Class java.lang.Object does not have a @DefaultCoder annotation.
Building a Coder from the fallback CoderProvider failed: Cannot provide coder for type java.lang.Object: [email protected] could not provide a Coder for type java.lang.Object: Cannot provide ProtoCoder because java.lang.Object is not a subclass of com.google.protobuf.Message; [email protected] could not provide a Coder for type java.lang.Object: Cannot provide SerializableCoder because java.lang.Object does not implement Serializable.
Building a Coder from the @DefaultCoder annotation failed: Class org.apache.beam.sdk.values.KV does not have a @DefaultCoder annotation.
at org.apache.beam.sdk.repackaged.com.google.common.base.Preconditions.checkState(Preconditions.java:444)
at org.apache.beam.sdk.values.TypedPValue.getCoder(TypedPValue.java:51)
at org.apache.beam.sdk.values.PCollection.getCoder(PCollection.java:130)
at org.apache.beam.sdk.values.TypedPValue.finishSpecifying(TypedPValue.java:90)
at org.apache.beam.sdk.runners.TransformHierarchy.finishSpecifyingInput(TransformHierarchy.java:143)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:418)
at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:334)
at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:154)
at org.apache.beam.sdk.transforms.Combine$Globally.expand(Combine.java:1459)
at org.apache.beam.sdk.transforms.Combine$Globally.expand(Combine.java:1336)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:420)
at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:350)
at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:167)
at ***************************.main(***************.java:231)
... 6 more
사과 :이 로컬로 사용 메이븐과 DirectRunner을 실행하려고 할 때
public static class CombineEventsFn extends CombineFn<Event, CombineEventsFn.Accum, EventListWrapper> {
public static class Accum {
EventListWrapper eventListWrapper = new EventListWrapper();
}
@Override
public Accum createAccumulator() {
return new Accum();
}
@Override
public Accum addInput(Accum accumulator, Event event) {
accumulator.eventListWrapper.events.add(event);
return accumulator;
}
@Override
public Accum mergeAccumulators(Iterable<Accum> accumulators) {
Accum merged = createAccumulator();
for (Accum accum : accumulators) {
merged.eventListWrapper.events.addAll(accum.eventListWrapper.events);
}
return merged;
}
@Override
public EventListWrapper extractOutput(Accum accumulator) {
return accumulator.eventListWrapper;
}
}
, 우리는 다음과 같은 오류를 얻고있다 덤프 - 모든 컨텍스트를 제공하려고합니다. 지금까지 내가 우리의 파이프 라인 String, Event 및 EventListWrapper 사이의 유형을 변화 말할 수있는 - - 후자의 두 클래스는 기본 코더가
나는대로 모두 java.lang.Object 및 org.apache.beam.sdk.values.KV<K, OutputT>에 대한 기본 코더에 대해 불평 왜 궁금 클래스 자체에 설정됩니다 (두 경우 모두 AvroCoder).
오류가 우리가 CombineFn을 적용한 라인에서 발생합니다 -이 변환을하지 않으면 파이프 라인이 작동하는지 확인할 수 있습니다.
우리는 결합 변환을 어떻게 든 잘못 설정했다고 생각하지만, Beam 설명서에서 올바른 방향으로 우리를 가리키는 내용을 찾지 못했습니다.
모든 통찰력은 감사 할 것입니다 - 미리 감사드립니다!
어떤 빔 버전을 사용하고 있습니까? –
죄송합니다. 아마도 다음을 포함해야합니다. 0.6.0 – cstaikos
은 Event.class 코더가 등록되어 있습니까? – CasualT