# pbspark
This package provides a way to convert protobuf messages into pyspark dataframes and vice versa using pyspark `udf`s.
## Installation
To install:
```bash
pip install pbspark
```
## Usage
Suppose we have a pyspark DataFrame which contains a column `value` which has protobuf encoded messages of our `SimpleMessage`:
```protobuf
syntax = "proto3";
package example;
message SimpleMessage {
string name = 1;
int64 quantity = 2;
float measure = 3;
}
```
### Basic conversion functions
There are two functions for operating on columns, `to_protobuf` and `from_protobuf`. These operations convert to/from an encoded protobuf column to a column of a struct representing the inferred message structure. `MessageConverter` instances (discussed below) can optionally be passed to these functions.
```python
from pyspark.sql.session import SparkSession
from example.example_pb2 import SimpleMessage
from pbspark import from_protobuf
from pbspark import to_protobuf
spark = SparkSession.builder.getOrCreate()
example = SimpleMessage(name="hello", quantity=5, measure=12.3)
data = [{"value": example.SerializeToString()}]
df_encoded = spark.createDataFrame(data)
df_decoded = df_encoded.select(from_protobuf(df_encoded.value, SimpleMessage).alias("value"))
df_expanded = df_decoded.select("value.*")
df_expanded.show()
# +-----+--------+-------+
# | name|quantity|measure|
# +-----+--------+-------+
# |hello| 5| 12.3|
# +-----+--------+-------+
df_reencoded = df_decoded.select(to_protobuf(df_decoded.value, SimpleMessage).alias("value"))
```
There are two helper functions, `df_to_protobuf` and `df_from_protobuf` for use on dataframes. They have a kwarg `expanded`, which will also take care of expanding/contracting the data between the single `value` column used in these examples and a dataframe which contains a column for each message field. `MessageConverter` instances (discussed below) can optionally be passed to these functions.
```python
from pyspark.sql.session import SparkSession
from example.example_pb2 import SimpleMessage
from pbspark import df_from_protobuf
from pbspark import df_to_protobuf
spark = SparkSession.builder.getOrCreate()
example = SimpleMessage(name="hello", quantity=5, measure=12.3)
data = [{"value": example.SerializeToString()}]
df_encoded = spark.createDataFrame(data)
# expanded=True will perform a `.select("value.*")` after converting,
# resulting in each protobuf field having its own column
df_expanded = df_from_protobuf(df_encoded, SimpleMessage, expanded=True)
df_expanded.show()
# +-----+--------+-------+
# | name|quantity|measure|
# +-----+--------+-------+
# |hello| 5| 12.3|
# +-----+--------+-------+
# expanded=True will first pack data using `struct([df[c] for c in df.columns])`,
# use this if the passed dataframe is already expanded
df_reencoded = df_to_protobuf(df_expanded, SimpleMessage, expanded=True)
```
### Column conversion using the `MessageConverter`
The four helper functions above are also available as methods on the `MessageConverter` class. Using an instance of `MessageConverter` we can decode the column of encoded messages into a column of spark `StructType` and then expand the fields.
```python
from pyspark.sql.session import SparkSession
from pbspark import MessageConverter
from example.example_pb2 import SimpleMessage
spark = SparkSession.builder.getOrCreate()
example = SimpleMessage(name="hello", quantity=5, measure=12.3)
data = [{"value": example.SerializeToString()}]
df_encoded = spark.createDataFrame(data)
mc = MessageConverter()
df_decoded = df_encoded.select(mc.from_protobuf(df_encoded.value, SimpleMessage).alias("value"))
df_expanded = df_decoded.select("value.*")
df_expanded.show()
# +-----+--------+-------+
# | name|quantity|measure|
# +-----+--------+-------+
# |hello| 5| 12.3|
# +-----+--------+-------+
df_expanded.schema
# StructType(List(StructField(name,StringType,true),StructField(quantity,IntegerType,true),StructField(measure,FloatType,true))
```
We can also re-encode them into protobuf.
```python
df_reencoded = df_decoded.select(mc.to_protobuf(df_decoded.value, SimpleMessage).alias("value"))
```
For expanded data, we can also encode after packing into a struct column:
```python
from pyspark.sql.functions import struct
df_unexpanded = df_expanded.select(
struct([df_expanded[c] for c in df_expanded.columns]).alias("value")
)
df_reencoded = df_unexpanded.select(
mc.to_protobuf(df_unexpanded.value, SimpleMessage).alias("value")
)
```
### Conversion details
Internally, `pbspark` uses protobuf's `MessageToDict`, which deserializes everything into JSON compatible objects by default. The exceptions are
* protobuf's bytes type, which `MessageToDict` would decode to a base64-encoded string; `pbspark` will decode any bytes fields directly to a spark `BinaryType`.
* protobuf's well known type, Timestamp type, which `MessageToDict` would decode to a string; `pbspark` will decode any Timestamp messages directly to a spark `TimestampType` (via python datetime objects).
* protobuf's int64 types, which `MessageToDict` would decode to a string for compatibility reasons; `pbspark` will decode these to `LongType`.
### Custom conversion of message types
Custom serde is also supported. Suppose we use our `NestedMessage` from the repository's example and we want to serialize the key and value together into a single string.
```protobuf
message NestedMessage {
string key = 1;
string value = 2;
}
```
We can create and register a custom serializer with the `MessageConverter`.
```python
from pbspark import MessageConverter
from example.example_pb2 import ExampleMessage
from example.example_pb2 import NestedMessage
from pyspark.sql.types import StringType
mc = MessageConverter()
# register a custom serializer
# this will serialize the NestedMessages into a string rather than a
# struct with `key` and `value` fields
encode_nested = lambda message: message.key + ":" + message.value
mc.register_serializer(NestedMessage, encode_nested, StringType())
# ...
from pyspark.sql.session import SparkSession
from pyspark import SparkContext
from pyspark.serializers import CloudPickleSerializer
sc = SparkContext(serializer=CloudPickleSerializer())
spark = SparkSession(sc).builder.getOrCreate()
message = ExampleMessage(nested=NestedMessage(key="hello", value="world"))
data = [{"value": message.SerializeToString()}]
df_encoded = spark.createDataFrame(data)
df_decoded = df_encoded.select(mc.from_protobuf(df_encoded.value, ExampleMessage).alias("value"))
# rather than a struct the value of `nested` is a string
df_decoded.select("value.nested").show()
# +-----------+
# | nested|
# +-----------+
# |hello:world|
# +-----------+
```
### How to write conversion functions
More generally, custom serde functions should be written in the following format.
```python
# Encoding takes a message instance and returns the result
# of the custom transformation.
def encode_nested(message: NestedMessage) -> str:
return message.key + ":" + message.value
# Decoding takes the encoded value, a message instance, and path string
# and populates the fields of the message instance. It returns `None`.
# The path str is used in the protobuf parser to log parse error info.
# Note that the first argument type should match the return type of the
# encoder if using both.
def decode_nested(s: str, message: NestedMessage, path: str):
key, value = s.split(":")
message.key = key
message.value = value
```
### Avoiding PicklingErrors
A seemingly common issue with protobuf and distributed processing is when a `PicklingError` is encountered when transmitting (pickling) protobuf message types from a main process to a fork. To avoid this, you need to ensure that the fully qualified module name in your protoc-generated python file is the same as the module path from which the message type is imported. In other words, for the example here, the descriptor module passed to the builder is `example.example_pb2`
```python
# from example/example_pb2.py
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "example.example_pb2", globals())
^^^^^^^^^^^^^^^^^^^
```
And to import the message type we would call the same module path:
```python
from example.example_pb2 import ExampleMessage
^^^^^^^^^^^^^^^^^^^
```
Note that the import module is the same as the one passed to the builder from the protoc-generated python. If these do not match, then you will encounter a `PicklingError`. From the pickle documentation: *pickle can save and restore class instances transparently, however the class definition must be importable and live in the same module as when the object was stored.*
To ensure that the module path is correct, you should run `protoc` from the relative root path of your proto files. For example, in this project, in the `Makefile` under the `gen` command, we call `protoc` from the project root rather than from within the `example` directory.
```makefile
export PROTO_PATH=.
gen:
poetry run protoc -I $$PROTO_PATH --python_out=$$PROTO_PATH --mypy_out=$$PROTO_PATH --proto_path=$$PROTO_PATH $$PROTO_PATH/example/*.proto
```
### Known issues
`RecursionError` when using self-referencing protobuf messages. Spark schemas do not allow for arbitrary depth, so protobuf messages which are circular- or self-referencing will result in infinite recursion errors when inferring the schema. If you have message structures like this you should resort to creating custom conversion functions, which forcibly limit the structural depth when converting these messages.
## Development
Ensure that [asdf](https://asdf-vm.com/) is installed, then run `make setup`.
* To format code `make fmt`
* To test code `make test`
* To run protoc `make gen`
Raw data
{
"_id": null,
"home_page": "https://github.com/crflynn/pbspark",
"name": "pbspark",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.7,<4.0",
"maintainer_email": "",
"keywords": "spark,protobuf,pyspark",
"author": "flynn",
"author_email": "crf204@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/80/ff/e22d1869bfb855613927cb4a1fb3ee8e0e362f4c85c416fd70e8f8f82fc6/pbspark-0.9.0.tar.gz",
"platform": null,
"description": "# pbspark\n\nThis package provides a way to convert protobuf messages into pyspark dataframes and vice versa using pyspark `udf`s.\n\n## Installation\n\nTo install:\n\n```bash\npip install pbspark\n```\n\n## Usage\n\nSuppose we have a pyspark DataFrame which contains a column `value` which has protobuf encoded messages of our `SimpleMessage`:\n\n```protobuf\nsyntax = \"proto3\";\n\npackage example;\n\nmessage SimpleMessage {\n string name = 1;\n int64 quantity = 2;\n float measure = 3;\n}\n```\n\n### Basic conversion functions\n\nThere are two functions for operating on columns, `to_protobuf` and `from_protobuf`. These operations convert to/from an encoded protobuf column to a column of a struct representing the inferred message structure. `MessageConverter` instances (discussed below) can optionally be passed to these functions.\n\n```python\nfrom pyspark.sql.session import SparkSession\nfrom example.example_pb2 import SimpleMessage\nfrom pbspark import from_protobuf\nfrom pbspark import to_protobuf\n\nspark = SparkSession.builder.getOrCreate()\n\nexample = SimpleMessage(name=\"hello\", quantity=5, measure=12.3)\ndata = [{\"value\": example.SerializeToString()}]\ndf_encoded = spark.createDataFrame(data)\n\ndf_decoded = df_encoded.select(from_protobuf(df_encoded.value, SimpleMessage).alias(\"value\"))\ndf_expanded = df_decoded.select(\"value.*\")\ndf_expanded.show()\n\n# +-----+--------+-------+\n# | name|quantity|measure|\n# +-----+--------+-------+\n# |hello| 5| 12.3|\n# +-----+--------+-------+\n\ndf_reencoded = df_decoded.select(to_protobuf(df_decoded.value, SimpleMessage).alias(\"value\"))\n```\n\nThere are two helper functions, `df_to_protobuf` and `df_from_protobuf` for use on dataframes. They have a kwarg `expanded`, which will also take care of expanding/contracting the data between the single `value` column used in these examples and a dataframe which contains a column for each message field. `MessageConverter` instances (discussed below) can optionally be passed to these functions.\n\n```python\nfrom pyspark.sql.session import SparkSession\nfrom example.example_pb2 import SimpleMessage\nfrom pbspark import df_from_protobuf\nfrom pbspark import df_to_protobuf\n\nspark = SparkSession.builder.getOrCreate()\n\nexample = SimpleMessage(name=\"hello\", quantity=5, measure=12.3)\ndata = [{\"value\": example.SerializeToString()}]\ndf_encoded = spark.createDataFrame(data)\n\n# expanded=True will perform a `.select(\"value.*\")` after converting,\n# resulting in each protobuf field having its own column\ndf_expanded = df_from_protobuf(df_encoded, SimpleMessage, expanded=True)\ndf_expanded.show()\n\n# +-----+--------+-------+\n# | name|quantity|measure|\n# +-----+--------+-------+\n# |hello| 5| 12.3|\n# +-----+--------+-------+\n\n# expanded=True will first pack data using `struct([df[c] for c in df.columns])`,\n# use this if the passed dataframe is already expanded\ndf_reencoded = df_to_protobuf(df_expanded, SimpleMessage, expanded=True)\n```\n\n### Column conversion using the `MessageConverter`\n\nThe four helper functions above are also available as methods on the `MessageConverter` class. Using an instance of `MessageConverter` we can decode the column of encoded messages into a column of spark `StructType` and then expand the fields.\n\n```python\nfrom pyspark.sql.session import SparkSession\nfrom pbspark import MessageConverter\nfrom example.example_pb2 import SimpleMessage\n\nspark = SparkSession.builder.getOrCreate()\n\nexample = SimpleMessage(name=\"hello\", quantity=5, measure=12.3)\ndata = [{\"value\": example.SerializeToString()}]\ndf_encoded = spark.createDataFrame(data)\n\nmc = MessageConverter()\ndf_decoded = df_encoded.select(mc.from_protobuf(df_encoded.value, SimpleMessage).alias(\"value\"))\ndf_expanded = df_decoded.select(\"value.*\")\ndf_expanded.show()\n\n# +-----+--------+-------+\n# | name|quantity|measure|\n# +-----+--------+-------+\n# |hello| 5| 12.3|\n# +-----+--------+-------+\n\ndf_expanded.schema\n# StructType(List(StructField(name,StringType,true),StructField(quantity,IntegerType,true),StructField(measure,FloatType,true))\n```\n\nWe can also re-encode them into protobuf.\n\n```python\ndf_reencoded = df_decoded.select(mc.to_protobuf(df_decoded.value, SimpleMessage).alias(\"value\"))\n```\n\nFor expanded data, we can also encode after packing into a struct column:\n\n```python\nfrom pyspark.sql.functions import struct\n\ndf_unexpanded = df_expanded.select(\n struct([df_expanded[c] for c in df_expanded.columns]).alias(\"value\")\n)\ndf_reencoded = df_unexpanded.select(\n mc.to_protobuf(df_unexpanded.value, SimpleMessage).alias(\"value\")\n)\n```\n\n### Conversion details\n\nInternally, `pbspark` uses protobuf's `MessageToDict`, which deserializes everything into JSON compatible objects by default. The exceptions are\n* protobuf's bytes type, which `MessageToDict` would decode to a base64-encoded string; `pbspark` will decode any bytes fields directly to a spark `BinaryType`.\n* protobuf's well known type, Timestamp type, which `MessageToDict` would decode to a string; `pbspark` will decode any Timestamp messages directly to a spark `TimestampType` (via python datetime objects).\n* protobuf's int64 types, which `MessageToDict` would decode to a string for compatibility reasons; `pbspark` will decode these to `LongType`.\n\n### Custom conversion of message types\n\nCustom serde is also supported. Suppose we use our `NestedMessage` from the repository's example and we want to serialize the key and value together into a single string.\n\n```protobuf\nmessage NestedMessage {\n string key = 1;\n string value = 2;\n}\n```\n\nWe can create and register a custom serializer with the `MessageConverter`.\n\n```python\nfrom pbspark import MessageConverter\nfrom example.example_pb2 import ExampleMessage\nfrom example.example_pb2 import NestedMessage\nfrom pyspark.sql.types import StringType\n\nmc = MessageConverter()\n\n# register a custom serializer\n# this will serialize the NestedMessages into a string rather than a\n# struct with `key` and `value` fields\nencode_nested = lambda message: message.key + \":\" + message.value\n\nmc.register_serializer(NestedMessage, encode_nested, StringType())\n\n# ...\n\nfrom pyspark.sql.session import SparkSession\nfrom pyspark import SparkContext\nfrom pyspark.serializers import CloudPickleSerializer\n\nsc = SparkContext(serializer=CloudPickleSerializer())\nspark = SparkSession(sc).builder.getOrCreate()\n\nmessage = ExampleMessage(nested=NestedMessage(key=\"hello\", value=\"world\"))\ndata = [{\"value\": message.SerializeToString()}]\ndf_encoded = spark.createDataFrame(data)\n\ndf_decoded = df_encoded.select(mc.from_protobuf(df_encoded.value, ExampleMessage).alias(\"value\"))\n# rather than a struct the value of `nested` is a string\ndf_decoded.select(\"value.nested\").show()\n\n# +-----------+\n# | nested|\n# +-----------+\n# |hello:world|\n# +-----------+\n```\n\n### How to write conversion functions\n\nMore generally, custom serde functions should be written in the following format.\n\n```python\n# Encoding takes a message instance and returns the result\n# of the custom transformation.\ndef encode_nested(message: NestedMessage) -> str:\n return message.key + \":\" + message.value\n\n# Decoding takes the encoded value, a message instance, and path string\n# and populates the fields of the message instance. It returns `None`.\n# The path str is used in the protobuf parser to log parse error info.\n# Note that the first argument type should match the return type of the\n# encoder if using both.\ndef decode_nested(s: str, message: NestedMessage, path: str):\n key, value = s.split(\":\")\n message.key = key\n message.value = value\n```\n\n### Avoiding PicklingErrors\n\nA seemingly common issue with protobuf and distributed processing is when a `PicklingError` is encountered when transmitting (pickling) protobuf message types from a main process to a fork. To avoid this, you need to ensure that the fully qualified module name in your protoc-generated python file is the same as the module path from which the message type is imported. In other words, for the example here, the descriptor module passed to the builder is `example.example_pb2`\n\n```python\n# from example/example_pb2.py\n_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, \"example.example_pb2\", globals())\n ^^^^^^^^^^^^^^^^^^^\n```\n\nAnd to import the message type we would call the same module path:\n\n```python\nfrom example.example_pb2 import ExampleMessage\n ^^^^^^^^^^^^^^^^^^^\n```\n\nNote that the import module is the same as the one passed to the builder from the protoc-generated python. If these do not match, then you will encounter a `PicklingError`. From the pickle documentation: *pickle can save and restore class instances transparently, however the class definition must be importable and live in the same module as when the object was stored.*\n\nTo ensure that the module path is correct, you should run `protoc` from the relative root path of your proto files. For example, in this project, in the `Makefile` under the `gen` command, we call `protoc` from the project root rather than from within the `example` directory.\n\n```makefile\nexport PROTO_PATH=.\n\ngen:\n\tpoetry run protoc -I $$PROTO_PATH --python_out=$$PROTO_PATH --mypy_out=$$PROTO_PATH --proto_path=$$PROTO_PATH $$PROTO_PATH/example/*.proto\n```\n\n### Known issues\n\n`RecursionError` when using self-referencing protobuf messages. Spark schemas do not allow for arbitrary depth, so protobuf messages which are circular- or self-referencing will result in infinite recursion errors when inferring the schema. If you have message structures like this you should resort to creating custom conversion functions, which forcibly limit the structural depth when converting these messages.\n\n## Development\n\nEnsure that [asdf](https://asdf-vm.com/) is installed, then run `make setup`.\n\n* To format code `make fmt`\n* To test code `make test`\n* To run protoc `make gen`\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Convert between protobuf messages and pyspark dataframes",
"version": "0.9.0",
"project_urls": {
"Documentation": "https://github.com/crflynn/pbspark",
"Homepage": "https://github.com/crflynn/pbspark",
"Repository": "https://github.com/crflynn/pbspark"
},
"split_keywords": [
"spark",
"protobuf",
"pyspark"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "c986193322e427c8c5287b12abea2ecf4d83091bb2bcf4954922c5a881e55bb1",
"md5": "b5052b9899b114e13070393699f2453e",
"sha256": "1faf944c14d18085f5f20e8f092fa2b35ad7dc922355eaee4765feb5b7f5904b"
},
"downloads": -1,
"filename": "pbspark-0.9.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b5052b9899b114e13070393699f2453e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7,<4.0",
"size": 10634,
"upload_time": "2023-06-07T23:22:04",
"upload_time_iso_8601": "2023-06-07T23:22:04.731934Z",
"url": "https://files.pythonhosted.org/packages/c9/86/193322e427c8c5287b12abea2ecf4d83091bb2bcf4954922c5a881e55bb1/pbspark-0.9.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "80ffe22d1869bfb855613927cb4a1fb3ee8e0e362f4c85c416fd70e8f8f82fc6",
"md5": "0c6e4a89d01730d2afc0c47e66fb7607",
"sha256": "c55e96673ceb6222a8035aaa071f330d546d38cf948320bfd39592017f869ff3"
},
"downloads": -1,
"filename": "pbspark-0.9.0.tar.gz",
"has_sig": false,
"md5_digest": "0c6e4a89d01730d2afc0c47e66fb7607",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7,<4.0",
"size": 13727,
"upload_time": "2023-06-07T23:22:06",
"upload_time_iso_8601": "2023-06-07T23:22:06.883874Z",
"url": "https://files.pythonhosted.org/packages/80/ff/e22d1869bfb855613927cb4a1fb3ee8e0e362f4c85c416fd70e8f8f82fc6/pbspark-0.9.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-06-07 23:22:06",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "crflynn",
"github_project": "pbspark",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pbspark"
}
JSON
