pyspark: 使用JavaObject StructType

Question

我需要解析一个JSON schema文件，以创建一个pyspark.sql.types.StructType。我找到了一个Scala库，可以帮助我完成这个任务。所以我会像这样调用它：

f = open('path/to/schema.json')
js = f.read()
conv = dspark.sparkContext._jvm.org.zalando.spark.jsonschema.SchemaConverter
schema = conv.convertContent(js)

但是当我尝试使用它来构建一个类似这样的DataFrame时：

spark.read.format("json").schema(schema)

我得到了以下错误：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/Cellar/apache-spark/2.1.0/libexec/python/pyspark/sql/readwriter.py", line 103, in schema
    raise TypeError("schema should be StructType")
TypeError: schema should be StructType

如果我打印类型：

print type(schema)

我得到：

<class 'py4j.java_gateway.JavaObject'>

如何将值包装为Python的StructType？

- Mark J Miller

Answer 1

在深入研究pyspark源代码后，我查看了DataFrame.schema的实现：

@property
@since(1.3)
def schema(self):
    if self._schema is None:
        try:
            self._schema = _parse_datatype_json_string(self._jdf.schema().json())
        except AttributeError as e:
            raise Exception(
                "Unable to parse datatype from schema. %s" % e)
    return self._schema

方法_parse_datatype_json_string定义在pyspark.sql.types中，所以这个工作：

from pyspark.sql.types import _parse_datatype_json_string

conv = self.spark.sparkContext._jvm.org.zalando.spark.jsonschema.SchemaConverter
jschema = conv.convertContent(read_schema)
schema = _parse_datatype_json_string(jschema.json())
src = src.schema(schema)

现在当我调用时：

print type(schema)

I get:

<class 'pyspark.sql.types.StructType'>

- Mark J Miller