我有一个 PTable<String, Pair<Entity1, Entity2>> 它是在运行转换作业的程序的中间阶段生成的。可接受条目示例：

["0067b4c054d14fe2-ACC8D37", 
 [{
    "unique_id": "0067b4c054d14fe2-ACC8D37",
    "user_id": "ACC8D37",
    "campaign_id": "3URL6GC2",
    "seller_id": "0067b4c"
}, {
    "unique_id": "0067b4c054d14fe2-ACC8D37",
    "user_id": "ACC8D37",
    "seller_id": "0067b4c"
}]]

我需要一个 PCollection<String> 哪里 Entity2 is null 变压器dofn

private PCollection<String> transformPairTable(PCollection<Pair<Entity1, Entity2>> pairPCollection) {
        return pairPCollection.parallelDo(new DoFn<Pair<Entity1, Entity2>, String>() {
            @Override
            public void process(Pair<Entity1, Entity2> entityPair, Emitter<String> emitter) {
                Entity1 entity1 = entityPair.first();
                Entity2 entity2 = entityPair.second();
                if (entity2 == null) {
                    String campaignId = entity1.getCampaignId();
                    emitter.emit(campaignId);
                }
            }
        }, Avros.strings());
    }

entity1和entity2都是从avro模式生成的类。
但当我运行作业时，它会引发运行时异常

Exception in thread "main" org.apache.crunch.CrunchRuntimeException: java.io.NotSerializableException: com.xxx.bb.data.vv.jobs.GenerateCount
    at org.apache.crunch.impl.mr.MRPipeline.plan(MRPipeline.java:140)
    at org.apache.crunch.impl.mr.MRPipeline.runAsync(MRPipeline.java:159)
    at org.apache.crunch.impl.mr.MRPipeline.run(MRPipeline.java:147)
    at org.apache.crunch.materialize.MaterializableIterable.iterator(MaterializableIterable.java:94)
    at org.apache.crunch.materialize.pobject.FirstElementPObject.process(FirstElementPObject.java:44)
    at org.apache.crunch.materialize.pobject.PObjectImpl.getValue(PObjectImpl.java:71)
    at com.xxxx.bb.xxxx.xxx.jobs.GenerateNewCount.getNewCustomersPCollection(GenerateNewCount.java:239)
    at com.xxx.x.data.x.jobs.GenerateNewCount.runJob(GenerateNewCount.java:78)
    at com.xxx.x.data.xx.jobs.BaseJob.run(BaseJob.java:87)
    at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:76)

我尝试了writeables.strings（），但它给出了相同的异常。
作业中使用的ptable具有entity2等于null的条目。
我试着用很多方法来转换ptable，但都不起作用。我想不出背后的主要原因。
当我使用

Entity1 p1= pairPCollection.first().getValue().first();

它引发以下异常：

ERROR materialize.MaterializableIterable: Could not materialize: Avro(/tmp/crunch-1893789994/p7)
java.io.IOException: No files found to materialize at: /tmp/crunch-1893789994/p7
    at org.apache.crunch.io.CompositePathIterable.create(CompositePathIterable.java:49)
    at org.apache.crunch.io.impl.FileSourceImpl.read(FileSourceImpl.java:134)
    at org.apache.crunch.io.avro.AvroFileSource.read(AvroFileSource.java:89)