如何查询具有复杂类型(如地图/数组)的RDD？例如,当我写这个测试代码时:
case class Test(name: String, map: Map[String, String])
val map = Map("hello" -> "world", "hey" -> "there")
val map2 = Map("hello" -> "people", "hey" -> "you")
val rdd = sc.parallelize(Array(Test("first", map), Test("second", map2)))
我虽然语法如下:
sqlContext.sql("SELECT * FROM rdd WHERE map.hello = world")
要么
sqlContext.sql("SELECT * FROM rdd WHERE map[hello] = world")
但我明白了
无法访问MapType类型中的嵌套字段(StringType,StringType,true)
和
org.apache.spark.sql.catalyst.errors.package $ TreeNodeException:未解析的属性
分别.
我有一个带有 MapType 字段的数据框。
>>> from pyspark.sql.functions import *
>>> from pyspark.sql.types import *
>>> fields = StructType([
... StructField('timestamp', TimestampType(), True),
... StructField('other_field', StringType(), True),
... StructField('payload', MapType(
... keyType=StringType(),
... valueType=StringType()),
... True), ])
>>> import datetime
>>> rdd = sc.parallelize([[datetime.datetime.now(), 'this should be in', {'akey': 'aValue'}]])
>>> df = rdd.toDF(fields)
>>> df.show()
+--------------------+-----------------+-------------------+
| timestamp| other_field| payload|
+--------------------+-----------------+-------------------+
|2018-01-10 12:56:...|this should be in|Map(akey -> aValue)|
+--------------------+-----------------+-------------------+
我想添加other_field作为字段中的键payload。
我知道我可以使用 udf：
>>> def _add_to_map(name, value, map_field): …我有一个数据框（在 Pyspark 中），其中一个行值作为字典：
df.show()
它看起来像：
+----+---+-----------------------------+
|name|age|info |
+----+---+-----------------------------+
|rob |26 |{color: red, car: volkswagen}|
|evan|25 |{color: blue, car: mazda} |
+----+---+-----------------------------+
根据评论给出更多：
df.printSchema()
类型是字符串
root
|-- name: string (nullable = true)
|-- age: string (nullable = true)
|-- dict: string (nullable = true)
是否可以从字典（颜色和汽车）中获取键并将它们设置为数据框中的列，并将值作为这些列的行？
预期结果：
+----+---+-----------------------------+
|name|age|color |car |
+----+---+-----------------------------+
|rob |26 |red |volkswagen |
|evan|25 |blue |mazda |
+----+---+-----------------------------+
我不知道我必须使用 df.withColumn() 并以某种方式遍历字典来选择每个字典，然后从中创建一列？到目前为止，我已经尝试找到一些答案，但大多数都使用 Pandas，而不是 Spark，所以我不确定是否可以应用相同的逻辑。
我有一个带有单列的 DataFrame，它是一个结构数组
df.printSchema()
root
|-- dataCells: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- label: string (nullable = true)
| | |-- value: string (nullable = true)
一些示例数据可能如下所示：
df.first()
Row(dataCells=[Row(label="firstName", value="John"), Row(label="lastName", value="Doe"), Row(label="Date", value="1/29/2018")])
我试图弄清楚如何通过将每个结构转换为命名列来重新格式化此 DataFrame。我想要一个像这样的数据框：
------------------------------------
| firstName | lastName | Date |
------------------------------------
| John | Doe | 1/29/2018 |
| .... | ... | ... |
我已经尝试了我能想到的一切，但还没有弄清楚。