如何将Pyspark Executor的Python模块拉开？

Question

我正在解析 protobuf 使用pyspark-streaming。
我的UDF是：

def parse_protobuf_from_bytes(pb_bytes):
    data = contacts_pb2.Contacts()
    data.ParseFromString(pb_bytes)
    return str(data)
parse_protobuf = udf(lambda x: parse_protobuf_from_bytes(x), StringType())

我正在调用代码：

df = spark.readStream \
            .format("kafka") \
            .options(**kafka_conf) \
            .load()

data = df.selectExpr("offset","CAST(key AS STRING)", "CAST(value AS STRING)") \
         .withColumn("event", parse_protobuf(col("value")))
df2 = data.select(col("event"))

df2.writeStream \
            .format("console") \
            .outputMode("append") \
            .option("truncate", False) \
            .start() \
            .awaitTermination()

我会遇到错误：

TID 0, spark.executor.node.com, executor 1): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/opt/cloudera/parcels/IMMUTA-2.8.1_20210803-6.3.4-spark2-public/python/pyspark/worker.py", line 366, in main
    func, profiler, deserializer, serializer = read_udfs(pickleSer, infile, eval_type)
  File "/opt/cloudera/parcels/IMMUTA-2.8.1_20210803-6.3.4-spark2-public/python/pyspark/worker.py", line 241, in read_udfs
    arg_offsets, udf = read_single_udf(pickleSer, infile, eval_type, runner_conf)
  File "/opt/cloudera/parcels/IMMUTA-2.8.1_20210803-6.3.4-spark2-public/python/pyspark/worker.py", line 168, in read_single_udf
    f, return_type = read_command(pickleSer, infile)
  File "/opt/cloudera/parcels/IMMUTA-2.8.1_20210803-6.3.4-spark2-public/python/pyspark/worker.py", line 69, in read_command
    command = serializer._read_with_length(file)
  File "/opt/cloudera/parcels/IMMUTA-2.8.1_20210803-6.3.4-spark2-public/python/pyspark/serializers.py", line 172, in _read_with_length
    return self.loads(obj)
  File "/opt/cloudera/parcels/IMMUTA-2.8.1_20210803-6.3.4-spark2-public/python/pyspark/serializers.py", line 580, in loads
    return pickle.loads(obj, encoding=encoding)
  File "/opt/cloudera/parcels/IMMUTA-2.8.1_20210803-6.3.4-spark2-public/python/pyspark/cloudpickle.py", line 875, in subimport
    __import__(name)
  File "/local/4/yarn/nm/usercache/myusername/appcache/application_1654889270563_0237/container_e119_1654889270563_0237_01_000002/__pyfiles__/contacts_pb2.py", line 5, in <module>
    from google.protobuf.internal import builder as _builder
ImportError: cannot import name 'builder'

        at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:456)
        at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:81)
        at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:64)
        at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:410)
        at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
        at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$11$$anon$1.hasNext(WholeStageCodegenExec.scala:624)
        at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$$anonfun$run$3.apply(WriteToDataSourceV2Exec.scala:117)
        at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$$anonfun$run$3.apply(WriteToDataSourceV2Exec.scala:116)
        at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1442)
        at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.run(WriteToDataSourceV2Exec.scala:146)
        at org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec$$anonfun$doExecute$2.apply(WriteToDataSourceV2Exec.scala:67)
        at org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec$$anonfun$doExecute$2.apply(WriteToDataSourceV2Exec.scala:66)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:121)
        at org.apache.spark.executor.Executor$TaskRunner$$anonfun$11.apply(Executor.scala:407)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1408)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:413)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

即使我在Spark Configuraion中传递软件包：

"spark.submit.pyFiles":"/home/myusername/testing_aiman/protofiles/google.zip,/home/myusername/testing_aiman/protofiles/compiled_protofiles/contacts_pb2.py"

我认为我没有正确包装Protobuf Zip文件。 我创建 google.zip 的方式是，从我的Conda-Environment的站点包装目录（/home/myusername/.conda/pkgs/protobuf-3.2.2-py36_0/ lib/python3.6/site-packages），我用google* z缩了所有内容。 我正在获取带有内容的ZIP文件：

(dev) [myusername@myedgenode protofiles]$ zip -sf google.zip
Archive contains:
  google/
  google/protobuf/
  google/protobuf/descriptor_pb2.py
  google/protobuf/timestamp_pb2.py
  google/protobuf/unittest_no_arena_import_pb2.py
  google/protobuf/any_pb2.py
  google/protobuf/field_mask_pb2.py
  google/protobuf/duration_pb2.py
  google/protobuf/empty_pb2.py
  google/protobuf/symbol_database.py
  google/protobuf/service.py
  google/protobuf/reflection.py
  google/protobuf/unittest_pb2.py
  google/protobuf/unittest_proto3_arena_pb2.py
  google/protobuf/pyext/
  google/protobuf/pyext/python_pb2.py
  google/protobuf/pyext/cpp_message.py
  google/protobuf/pyext/__init__.py
  google/protobuf/pyext/__pycache__/
  google/protobuf/pyext/__pycache__/__init__.cpython-36.pyc
  google/protobuf/pyext/__pycache__/cpp_message.cpython-36.pyc
  google/protobuf/pyext/__pycache__/python_pb2.cpython-36.pyc
  google/protobuf/pyext/_message.cpython-36m-x86_64-linux-gnu.so
  google/protobuf/struct_pb2.py
  google/protobuf/any_test_pb2.py
  google/protobuf/source_context_pb2.py
  google/protobuf/__init__.py
  google/protobuf/message.py
  google/protobuf/unittest_import_pb2.py
  google/protobuf/text_format.py
  google/protobuf/internal/
  google/protobuf/internal/test_util.py
  google/protobuf/internal/well_known_types.py
  google/protobuf/internal/text_format_test.py
  google/protobuf/internal/message_set_extensions_pb2.py
  google/protobuf/internal/descriptor_pool_test2_pb2.py
  google/protobuf/internal/well_known_types_test.py
  google/protobuf/internal/factory_test1_pb2.py
  google/protobuf/internal/enum_type_wrapper.py
  google/protobuf/internal/generator_test.py
  google/protobuf/internal/_api_implementation.cpython-36m-x86_64-linux-gnu.so
  google/protobuf/internal/missing_enum_values_pb2.py
  google/protobuf/internal/descriptor_database_test.py
  google/protobuf/internal/decoder.py
  google/protobuf/internal/more_messages_pb2.py
  google/protobuf/internal/descriptor_pool_test1_pb2.py
  google/protobuf/internal/more_extensions_dynamic_pb2.py
  google/protobuf/internal/any_test_pb2.py
  google/protobuf/internal/more_extensions_pb2.py
  google/protobuf/internal/test_bad_identifiers_pb2.py
  google/protobuf/internal/json_format_test.py
  google/protobuf/internal/__init__.py
  google/protobuf/internal/wire_format_test.py
  google/protobuf/internal/symbol_database_test.py
  google/protobuf/internal/service_reflection_test.py
  google/protobuf/internal/factory_test2_pb2.py
  google/protobuf/internal/python_message.py
  google/protobuf/internal/descriptor_test.py
  google/protobuf/internal/proto_builder_test.py
  google/protobuf/internal/__pycache__/
  google/protobuf/internal/__pycache__/test_util.cpython-36.pyc
  google/protobuf/internal/__pycache__/service_reflection_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/testing_refleaks.cpython-36.pyc
  google/protobuf/internal/__pycache__/factory_test1_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/text_encoding_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/wire_format.cpython-36.pyc
  google/protobuf/internal/__pycache__/text_format_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/any_test_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/descriptor_pool_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/more_extensions_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/reflection_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/more_messages_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/proto_builder_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/descriptor_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/descriptor_pool_test1_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/decoder.cpython-36.pyc
  google/protobuf/internal/__pycache__/__init__.cpython-36.pyc
  google/protobuf/internal/__pycache__/file_options_test_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/well_known_types_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/python_message.cpython-36.pyc
  google/protobuf/internal/__pycache__/descriptor_pool_test2_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/enum_type_wrapper.cpython-36.pyc
  google/protobuf/internal/__pycache__/more_extensions_dynamic_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/type_checkers.cpython-36.pyc
  google/protobuf/internal/__pycache__/factory_test2_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/message_set_extensions_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/test_bad_identifiers_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/descriptor_database_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/message_listener.cpython-36.pyc
  google/protobuf/internal/__pycache__/containers.cpython-36.pyc
  google/protobuf/internal/__pycache__/symbol_database_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/encoder.cpython-36.pyc
  google/protobuf/internal/__pycache__/message_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/json_format_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/_parameterized.cpython-36.pyc
  google/protobuf/internal/__pycache__/unknown_fields_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/well_known_types.cpython-36.pyc
  google/protobuf/internal/__pycache__/wire_format_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/api_implementation.cpython-36.pyc
  google/protobuf/internal/__pycache__/missing_enum_values_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/packed_field_test_pb2.cpython-36.pyc
  google/protobuf/internal/__pycache__/message_factory_test.cpython-36.pyc
  google/protobuf/internal/__pycache__/generator_test.cpython-36.pyc
  google/protobuf/internal/type_checkers.py
  google/protobuf/internal/reflection_test.py
  google/protobuf/internal/encoder.py
  google/protobuf/internal/message_listener.py
  google/protobuf/internal/descriptor_pool_test.py
  google/protobuf/internal/message_factory_test.py
  google/protobuf/internal/_parameterized.py
  google/protobuf/internal/api_implementation.py
  google/protobuf/internal/containers.py
  google/protobuf/internal/wire_format.py
  google/protobuf/internal/message_test.py
  google/protobuf/internal/text_encoding_test.py
  google/protobuf/internal/packed_field_test_pb2.py
  google/protobuf/internal/unknown_fields_test.py
  google/protobuf/internal/file_options_test_pb2.py
  google/protobuf/internal/import_test_package/
  google/protobuf/internal/import_test_package/outer_pb2.py
  google/protobuf/internal/import_test_package/inner_pb2.py
  google/protobuf/internal/import_test_package/__init__.py
  google/protobuf/internal/import_test_package/__pycache__/
  google/protobuf/internal/import_test_package/__pycache__/__init__.cpython-36.pyc
  google/protobuf/internal/import_test_package/__pycache__/outer_pb2.cpython-36.pyc
  google/protobuf/internal/import_test_package/__pycache__/inner_pb2.cpython-36.pyc
  google/protobuf/internal/testing_refleaks.py
  google/protobuf/unittest_no_generic_services_pb2.py
  google/protobuf/json_format.py
  google/protobuf/api_pb2.py
  google/protobuf/__pycache__/
  google/protobuf/__pycache__/wrappers_pb2.cpython-36.pyc
  google/protobuf/__pycache__/unittest_import_public_pb2.cpython-36.pyc
  google/protobuf/__pycache__/type_pb2.cpython-36.pyc
  google/protobuf/__pycache__/descriptor.cpython-36.pyc
  google/protobuf/__pycache__/descriptor_pb2.cpython-36.pyc
  google/protobuf/__pycache__/message.cpython-36.pyc
  google/protobuf/__pycache__/unittest_custom_options_pb2.cpython-36.pyc
  google/protobuf/__pycache__/test_messages_proto3_pb2.cpython-36.pyc
  google/protobuf/__pycache__/any_test_pb2.cpython-36.pyc
  google/protobuf/__pycache__/any_pb2.cpython-36.pyc
  google/protobuf/__pycache__/proto_builder.cpython-36.pyc
  google/protobuf/__pycache__/service.cpython-36.pyc
  google/protobuf/__pycache__/text_format.cpython-36.pyc
  google/protobuf/__pycache__/reflection.cpython-36.pyc
  google/protobuf/__pycache__/descriptor_pool.cpython-36.pyc
  google/protobuf/__pycache__/__init__.cpython-36.pyc
  google/protobuf/__pycache__/unittest_pb2.cpython-36.pyc
  google/protobuf/__pycache__/duration_pb2.cpython-36.pyc
  google/protobuf/__pycache__/service_reflection.cpython-36.pyc
  google/protobuf/__pycache__/field_mask_pb2.cpython-36.pyc
  google/protobuf/__pycache__/unittest_proto3_arena_pb2.cpython-36.pyc
  google/protobuf/__pycache__/api_pb2.cpython-36.pyc
  google/protobuf/__pycache__/unittest_import_pb2.cpython-36.pyc
  google/protobuf/__pycache__/unittest_arena_pb2.cpython-36.pyc
  google/protobuf/__pycache__/unittest_mset_wire_format_pb2.cpython-36.pyc
  google/protobuf/__pycache__/unittest_no_arena_pb2.cpython-36.pyc
  google/protobuf/__pycache__/message_factory.cpython-36.pyc
  google/protobuf/__pycache__/timestamp_pb2.cpython-36.pyc
  google/protobuf/__pycache__/json_format.cpython-36.pyc
  google/protobuf/__pycache__/source_context_pb2.cpython-36.pyc
  google/protobuf/__pycache__/empty_pb2.cpython-36.pyc
  google/protobuf/__pycache__/map_unittest_pb2.cpython-36.pyc
  google/protobuf/__pycache__/symbol_database.cpython-36.pyc
  google/protobuf/__pycache__/struct_pb2.cpython-36.pyc
  google/protobuf/__pycache__/unittest_no_generic_services_pb2.cpython-36.pyc
  google/protobuf/__pycache__/descriptor_database.cpython-36.pyc
  google/protobuf/__pycache__/unittest_no_arena_import_pb2.cpython-36.pyc
  google/protobuf/__pycache__/text_encoding.cpython-36.pyc
  google/protobuf/__pycache__/unittest_mset_pb2.cpython-36.pyc
  google/protobuf/service_reflection.py
  google/protobuf/unittest_import_public_pb2.py
  google/protobuf/descriptor.py
  google/protobuf/unittest_arena_pb2.py
  google/protobuf/map_unittest_pb2.py
  google/protobuf/unittest_custom_options_pb2.py
  google/protobuf/type_pb2.py
  google/protobuf/unittest_no_arena_pb2.py
  google/protobuf/unittest_mset_pb2.py
  google/protobuf/test_messages_proto3_pb2.py
  google/protobuf/message_factory.py
  google/protobuf/descriptor_database.py
  google/protobuf/descriptor_pool.py
  google/protobuf/wrappers_pb2.py
  google/protobuf/unittest_mset_wire_format_pb2.py
  google/protobuf/text_encoding.py
  google/protobuf/proto_builder.py
Total 187 entries (5727926 bytes)

我是否必须在google/内部cd ，然后将Google> Google文件夹中的内容缩回？

Answer 1

通过安装Protobuf 版本-3.20。
此版本3.3没有文件 google/protobuf/internal/builder.py 。