相关疑难解决方法(0)

如何查询具有复杂类型(如地图/数组)的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:未解析的属性

分别.

我正在使用pyspark,使用spark-csv将大型csv文件加载到数据框中,作为预处理步骤,我需要对其中一列(包含json字符串)中可用的数据应用各种操作.这将返回X值,每个值都需要存储在各自独立的列中.

该功能将在UDF中实现.但是,我不确定如何从该UDF返回值列表并将这些值提供给单个列.下面是一个简单的例子:

(...)
from pyspark.sql.functions import udf
def udf_test(n):
    return [n/2, n%2]

test_udf=udf(udf_test)


df.select('amount','trans_date').withColumn("test", test_udf("amount")).show(4)

这产生以下结果:

+------+----------+--------------------+
|amount|trans_date|                test|
+------+----------+--------------------+
|  28.0|2016-02-07|         [14.0, 0.0]|
| 31.01|2016-02-07|[15.5050001144409...|
| 13.41|2016-02-04|[6.70499992370605...|
| 307.7|2015-02-17|[153.850006103515...|
| 22.09|2016-02-05|[11.0450000762939...|
+------+----------+--------------------+
only showing top 5 rows

将udf在不同的列上返回的两个值(在此示例中)存储的最佳方法是什么？现在他们被键入字符串:

df.select('amount','trans_date').withColumn("test", test_udf("amount")).printSchema()

root
 |-- amount: float (nullable = true)
 |-- trans_date: string (nullable = true)
 |-- test: string (nullable = true)

我有一个镶木地板文件中的数据有2个字段:object_id: String和alpha: Map<>.

它被读入sparkSQL中的数据框,模式如下所示:

scala> alphaDF.printSchema()
root
 |-- object_id: string (nullable = true)
 |-- ALPHA: map (nullable = true)
 |    |-- key: string
 |    |-- value: struct (valueContainsNull = true)

我正在使用Spark 2.0,我正在尝试创建一个新的数据框,其中列需要是地图的object_id加号键,ALPHA如object_id, key1, key2, key2, ...

我是第一次尝试看看我是否至少可以像这样访问地图:

scala> alphaDF.map(a => a(0)).collect()
<console>:32: error: Unable to find encoder for type stored in a Dataset.
Primitive types (Int, String, etc) and Product types (case classes) are 
supported by importing spark.implicits._  Support for serializing …

我的问题是如何将列拆分为多列.我不知道为什么df.toPandas()不起作用.

例如,我想将'df_test'更改为'df_test2'.我看到很多使用pandas模块的例子.还有另外一种方法吗？先感谢您.

df_test = sqlContext.createDataFrame([
(1, '14-Jul-15'),
(2, '14-Jun-15'),
(3, '11-Oct-15'),
], ('id', 'date'))

df_test2

id     day    month    year
1       14     Jul      15
2       14     Jun      15
1       11     Oct      15