小编xen*_*yon的帖子

我正在试图找出在Spark数据帧列中获得最大值的最佳方法.

请考虑以下示例:

df = spark.createDataFrame([(1., 4.), (2., 5.), (3., 6.)], ["A", "B"])
df.show()

这创造了:

+---+---+
|  A|  B|
+---+---+
|1.0|4.0|
|2.0|5.0|
|3.0|6.0|
+---+---+

我的目标是找到A列中的最大值(通过检查,这是3.0).使用PySpark,我可以想到以下四种方法:

# Method 1: Use describe()
float(df.describe("A").filter("summary = 'max'").select("A").first().asDict()['A'])

# Method 2: Use SQL
df.registerTempTable("df_table")
spark.sql("SELECT MAX(A) as maxval FROM df_table").first().asDict()['maxval']

# Method 3: Use groupby()
df.groupby().max('A').first().asDict()['max(A)']

# Method 4: Convert to RDD
df.select("A").rdd.max()[0]

上面的每一个都给出了正确的答案,但在没有Spark分析工具的情况下,我无法分辨哪个是最好的.

任何关于上述哪种方法在Spark运行时或资源使用方面最有效的直觉或经验主义的想法,或者是否有比上述方法更直接的方法？

是否有Hive查询可以快速查找表大小(即行数)而无需启动耗时的MapReduce作业？(这就是我想避免的原因COUNT(*).)

我试过了DESCRIBE EXTENDED,但那产生了numRows=0显然是不正确的.

(对于newb问题道歉.我尝试使用Google搜索并搜索apache.org文档但没有成功.)

我很难实现看起来应该很容易的东西:

我的目标是使用第二个RDD /数据帧作为查找表或翻译字典在RDD /数据帧中进行翻译.我想在多列中进行这些翻译.

解释问题的最简单方法是通过示例.假设我输入以下两个RDD:

Route SourceCityID DestinationCityID
A     1            2
B     1            3
C     2            1

和

CityID CityName
1      London
2      Paris
3      Tokyo

我想要的输出RDD是:

Route SourceCity DestinationCity
A     London     Paris
B     London     Tokyo
C     Paris      London

我应该如何制作呢？

这是SQL中的一个简单问题,但我不知道Spark中有明显的RDD解决方案.该加入,协同组等方法似乎不是很适合多列RDDS,不允许指定列加入上.

有任何想法吗？SQLContext是答案吗？

一个(Python)示例将使我的问题清楚.假设我有一个Spark数据框,其中包含在特定日期观看某些电影的人,如下所示:

movierecord = spark.createDataFrame([("Alice", 1, ["Avatar"]),("Bob", 2, ["Fargo", "Tron"]),("Alice", 4, ["Babe"]), ("Alice", 6, ["Avatar", "Airplane"]), ("Alice", 7, ["Pulp Fiction"]), ("Bob", 9, ["Star Wars"])],["name","unixdate","movies"])

上面定义的模式和数据框如下所示:

root
 |-- name: string (nullable = true)
 |-- unixdate: long (nullable = true)
 |-- movies: array (nullable = true)
 |    |-- element: string (containsNull = true)

+-----+--------+------------------+
|name |unixdate|movies            |
+-----+--------+------------------+
|Alice|1       |[Avatar]          |
|Bob  |2       |[Fargo, Tron]     |
|Alice|4       |[Babe]            |
|Alice|6       |[Avatar, Airplane]|
|Alice|7       |[Pulp Fiction]    |
|Bob  |9       |[Star Wars]       |
+-----+--------+------------------+

我想从上面开始生成一个新的数据帧列,其中包含 …

我正在寻找的:一种在Python中实现特殊乘法运算的方法,该矩阵恰好采用scipy稀疏(csr)格式.这是一种特殊的乘法,不是矩阵乘法,也不是Kronecker乘法,也不是Hadamard又是逐点乘法,并且在scipy.sparse中似乎没有任何内置支持.

所需操作:输出的每一行应包含两个输入矩阵中相应行元素的每个乘积的结果.所以开始具有两个相同尺寸的矩阵,每一个尺寸米由Ñ,结果应具有的尺寸米由N ^ 2.

它看起来像这样:

Python代码:

import scipy.sparse
A = scipy.sparse.csr_matrix(np.array([[1,2],[3,4]]))
B = scipy.sparse.csr_matrix(np.array([[0,5],[6,7]]))

# C is the desired product of A and B. It should look like:
C = scipy.sparse.csr_matrix(np.array([[0,5,0,10],[18,21,24,28]]))

什么是一个很好或有效的方法来做到这一点？我试过看看stackoverflow以及其他地方,到目前为止没有运气.到目前为止,听起来我最好的选择是在for循环中逐行操作,但这听起来很可怕,因为我的输入矩阵有几百万行和几千列,大多数是稀疏的.

由于Spark的mllib不具有最近邻居功能，因此我尝试将Annoy用于近似最近邻居。我尝试广播Annoy对象并将其传递给工作人员。但是，它没有按预期运行。

以下是可再现性的代码（将在PySpark中运行）。在将Annoy与不带Spark搭配使用时，看到的差异突出了问题。

from annoy import AnnoyIndex
import random
random.seed(42)

f = 40
t = AnnoyIndex(f)  # Length of item vector that will be indexed
allvectors = []
for i in xrange(20):
    v = [random.gauss(0, 1) for z in xrange(f)]
    t.add_item(i, v)
    allvectors.append((i, v))
t.build(10) # 10 trees

# Use Annoy with Spark
sparkvectors = sc.parallelize(allvectors)
bct = sc.broadcast(t)
x = sparkvectors.map(lambda x: bct.value.get_nns_by_vector(vector=x[1], n=5))
print "Five closest neighbors for first vector with Spark:",
print x.first()

# Use Annoy …