问使用自定义扩展服务运行梁时未知编码器URN beam:coder:pickled_python:v1

EN

def run():
      """Main function that defines pipeline and runs it."""
      pipeline_options = get_pipeline_options(**vars(args))

      request = ListDocumentsRequest()
      request.parent = 'projects/PROJECT_ID/databases'
      request.collection_id = 'pokemon'

      with beam.Pipeline(options=pipeline_options) as pipeline:
        _ = (pipeline
            | 'Create Requests' >> beam.Create([request.transaction])
                .with_output_types(typing.List[bytes])
            | 'Read from JavaFirestore' >> beam.ExternalTransform(
                'my.beam.transform.firestore_list_documents',
                ImplicitSchemaPayloadBuilder({'parent': request.parent,
                                            'collectionId': request.collection_id}),
                "localhost:12345")
            | 'Write' >> WriteToText('output_files/output')
            )

public static class FirestoreListDocumentsBuilder implements
      ExternalTransformBuilder<FirestoreTransformsConfiguration, PCollection<ListDocumentsRequest>, PCollection<Document>> {
  
    @Override
    public PTransform<PCollection<ListDocumentsRequest>, PCollection<Document>> buildExternal(
      FirestoreTransformsConfiguration configuration) {
      return FirestoreIO.v1().read().listDocuments().build();
    }
  }

@AutoService(ExternalTransformRegistrar.class)
public class FirestoreTransformsRegistrar implements ExternalTransformRegistrar {

  final static String URN_LIST_DOCS = "my.beam.transform.firestore_list_documents";  

  @Override
  public Map<String, ExternalTransformBuilder<?, ?, ?>> knownBuilderInstances() {
    return ImmutableMap.of(
      URN_LIST_DOCS,new FirestoreListDocumentsBuilder()
    );
  }
}

java.lang.IllegalArgumentException: Unknown Coder URN beam:coder:pickled_python:v1. Known URNs: [beam:coder:avro:generic:v1, beam:coder:bytes:v1, beam:coder:bool:v1, beam:coder:string_utf8:v1, beam:coder:kv:v1, beam:coder:varint:v1, beam:coder:interval_window:v1, beam:coder:iterable:v1, beam:coder:timer:v1, beam:coder:length_prefix:v1, beam:coder:global_window:v1, beam:coder:windowed_value:v1, beam:coder:param_windowed_value:v1, beam:coder:double:v1, beam:coder:row:v1, beam:coder:sharded_key:v1, beam:coder:custom_window:v1]