- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我有一个 df1
Spark 数据帧
id transactions
1 [1, 2, 3, 5]
2 [1, 2, 3, 6]
3 [1, 2, 9, 8]
4 [1, 2, 5, 6]
root
|-- id: int (nullable = true)
|-- transactions: array (nullable = false)
|-- element: int(containsNull = true)
None
我有一个 df2
Spark 数据帧
items cost
[1] 1.0
[2] 1.0
[2, 1] 2.0
[6, 1] 2.0
root
|-- items: array (nullable = false)
|-- element: int (containsNull = true)
|-- cost: int (nullable = true)
None
我想检查 items 列中的所有数组元素是否都在 transactions 列中。
第一行 ([1, 2, 3, 5]
) 包含项目列中的 [1],[2],[2, 1]
。因此我需要总结它们相应的成本:1.0 + 1.0 + 2.0 = 4.0
我想要的输出是
id transactions score
1 [1, 2, 3, 5] 4.0
2 [1, 2, 3, 6] 6.0
3 [1, 2, 9, 8] 4.0
4 [1, 2, 5, 6] 6.0
我尝试使用带有 collect()
/toLocalIterator
的循环,但它似乎效率不高。我会有大数据。
我认为创建这样的 UDF 可以解决问题。但它会引发错误。
from pyspark.sql.functions import udf
def containsAll(x, y):
result = all(elem in x for elem in y)
if result:
print("Yes, transactions contains all items")
else :
print("No")
contains_udf = udf(containsAll)
dataFrame.withColumn("result", contains_udf(df2.items, df1.transactions)).show()
还有其他办法吗?
最佳答案
2.4之前有效的udf(注意它必须返回一些东西
from pyspark.sql.functions import udf
@udf("boolean")
def contains_all(x, y):
if x is not None and y is not None:
return set(y).issubset(set(x))
在 2.4 或更高版本中不需要 udf:
from pyspark.sql.functions import array_intersect, size
def contains_all(x, y):
return size(array_intersect(x, y)) == size(y)
用法:
from pyspark.sql.functions import col, sum as sum_, when
df1 = spark.createDataFrame(
[(1, [1, 2, 3, 5]), (2, [1, 2, 3, 6]), (3, [1, 2, 9, 8]), (4, [1, 2, 5, 6])],
("id", "transactions")
)
df2 = spark.createDataFrame(
[([1], 1.0), ([2], 1.0), ([2, 1], 2.0), ([6, 1], 2.0)],
("items", "cost")
)
(df1
.crossJoin(df2).groupBy("id", "transactions")
.agg(sum_(when(
contains_all("transactions", "items"), col("cost")
)).alias("score"))
.show())
结果:
+---+------------+-----+
| id|transactions|score|
+---+------------+-----+
| 1|[1, 2, 3, 5]| 4.0|
| 4|[1, 2, 5, 6]| 6.0|
| 2|[1, 2, 3, 6]| 6.0|
| 3|[1, 2, 9, 8]| 4.0|
+---+------------+-----+
如果 df2
很小,最好将其用作局部变量:
items = sc.broadcast([
(set(items), cost) for items, cost in df2.select("items", "cost").collect()
])
def score(y):
@udf("double")
def _(x):
if x is not None:
transactions = set(x)
return sum(
cost for items, cost in y.value
if items.issubset(transactions))
return _
df1.withColumn("score", score(items)("transactions")).show()
+---+------------+-----+
| id|transactions|score|
+---+------------+-----+
| 1|[1, 2, 3, 5]| 4.0|
| 2|[1, 2, 3, 6]| 6.0|
| 3|[1, 2, 9, 8]| 4.0|
| 4|[1, 2, 5, 6]| 6.0|
+---+------------+-----+
终于可以爆破加入了
from pyspark.sql.functions import explode
costs = (df1
# Explode transactiosn
.select("id", explode("transactions").alias("item"))
.join(
df2
# Add id so we can later use it to identify source
.withColumn("_id", monotonically_increasing_id().alias("_id"))
# Explode items
.select(
"_id", explode("items").alias("item"),
# We'll need size of the original items later
size("items").alias("size"), "cost"),
["item"])
# Count matches in groups id, items
.groupBy("_id", "id", "size", "cost")
.count()
# Compute cost
.groupBy("id")
.agg(sum_(when(col("size") == col("count"), col("cost"))).alias("score")))
costs.show()
+---+-----+
| id|score|
+---+-----+
| 1| 4.0|
| 3| 4.0|
| 2| 6.0|
| 4| 6.0|
+---+-----+
然后将结果与原来的df1
连接起来,
df1.join(costs, ["id"])
但这并不是一个简单的解决方案,需要多次洗牌。它可能仍然优于笛卡尔积 (crossJoin
),但它取决于实际数据。
关于arrays - 检查另一个数组中存在的数组的所有元素,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54894115/
SELECT *, `o_cheque_request.member_id`, `o_cheque_request.wallet_id` FROM `o_cheque_request`, `o_mem
根据某一条件从数据库表中查询 『有』与『没有』,只有两种状态,那为什么在写SQL的时候,还要**SELECT count(*)**呢? 无论是刚入道的程序员新星,还是精湛沙场多年的程序员老白,都是一如
我试图找出一个文件是否存在,如果存在,验证css样式是否已经存在,如果不存在,将它们写在文件末尾... 我已经完成了这一切,但分 3 个步骤: 该文件是否存在? FileInfo fi= new Fi
我们正在开发即时消息传递应用程序,并且需要在用户的化身上用绿点显示用户 friend 的“状态”。 “状态”远远超出了“my_app_is_opened_and_on_focus”,这意味着(我猜可能
模式 Movie(title, year, director, budget, earnings) Actor(stagename, realname, birthyear) ActedIn(stag
我有一个正在尝试创建的 MySQL 触发器,但无法获得正确的语法。 触发器应该遍历一组关键字并将其与插入数据库的新帖子的标题进行匹配。如果找到匹配项,它应该将新帖子分配给该存储桶并更新存储桶的关键字集
我有 3 个表......用户、更新和碰撞。 我想向发出 api 请求的用户返回最新订单的 feed 更新,并提供显示 feed 中每个状态所需的所有数据。我还需要包括更新是否已被发出 api 请求的
我正在尝试呈现一个带有 UIView 的 UIViewController。 以下是我在 viewDidLoad 方法中尝试的代码。 //create the view controller UIVi
我正在努力弄清楚如何在不对 mysql 进行两次调用的情况下从一个表中检查两件事。 我有一个 Members 表。我想测试MemberID 列中是否存在某个值,以及PhoneNumber 列中是否存在
以下代码给出了一个没有 Do Compile 错误的循环: Loop Sheets("Snap").Rows(1).AutoFilter Field:=5, Criteria1:=List
是否可以通过检查“dig”的输出来检查域名的存在? 在绑定(bind)源中,我发现了这些常量: 0 DNS_R_NOEROR 1 DNS_R_FORMERR 2 DNS_R_SERVFAIL 3 DN
Controller 有问题 我在 Windows 上使用服务器,一切正常,但在互联网上我试图访问页面 social_apartament/beauty_life/并且找不到该页面,代码错误 404这
/** This is struct S. */ struct S(T) { static if(isFloatingPoint!T) { /// This version works
JVM 类型删除如何帮助 Clojure?没有它,Clojure 还能存在吗?如果 JVM 有具体化的类型会发生什么?也就是说,Clojure 将如何改变? 最佳答案 Clojure 根本不会有太大变
许多论文等提到对“system()”的调用是不安全且不可移植的。我不反对他们的论点。 不过,我注意到许多 Unix 实用程序都有一个等效的 C 库。如果没有,源可用于各种这些工具。 虽然许多论文和此类
在我的 Node js 应用程序中,我有一个用户登录 api。上面我在服务器端代码中创建了一个名为 customerid 的变量。现在,当用户身份验证成功时。我将他的 userid 值存储在我的 cu
我有一个工作资源管理器组,由 Ubuntu 14.04 虚拟机、网络接口(interface)、公共(public) IP 地址和存储帐户组成。我已经从这组资源中创建了一个模板。 当我尝试部署这组资源
我有一个函数createminor4(arr,锦标赛)它基本上将arr分成4组,每组8人,然后将它们一次交换到tourney 1组。从那里它插入四个{},其中有 4 个带有空数组的键。 我已经在 Ch
我有一个图表,其中有两个图例。我需要更改其中一个图例的点的大小。 我需要更改图例中“市场类型”的项目符号大小。我使用示例 here但不适用于我的图表。 我的代码如下: k <- ggplot(subs
我有 fiddle here展示我正在尝试做的事情。 我有一个动态生成的表,因此列可以按用户选择的任何顺序显示。因此,我尝试获取两个特定 header 的索引,以便可以将 CSS 类添加到这两列以供稍
我是一名优秀的程序员,十分优秀!