使用pyspark从每个行的数组中获取不同的计数
可以通过以下步骤实现：
- 导入必要的模块和函数：
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, col, countDistinct- 创建SparkSession对象：
spark = SparkSession.builder.appName("ArrayCount").getOrCreate()- 创建包含数组的DataFrame：
data = [("A", [1, 2, 3]),
("B", [2, 3, 4]),
("C", [3, 4, 5])]
df = spark.createDataFrame(data, ["id", "array_col"])- 使用explode函数将数组展开为多行：
df_exploded = df.select("id", explode("array_col").alias("value"))- 使用groupBy和countDistinct函数对每个行的数组元素进行计数：
result = df_exploded.groupBy("id").agg(countDistinct("value").alias("distinct_count"))- 打印结果：
result.show()完整的代码示例：
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, col, countDistinct
spark = SparkSession.builder.appName("ArrayCount").getOrCreate()
data = [("A", [1, 2, 3]),
("B", [2, 3, 4]),
("C", [3, 4, 5])]
df = spark.createDataFrame(data, ["id", "array_col"])
df_exploded = df.select("id", explode("array_col").alias("value"))
result = df_exploded.groupBy("id").agg(countDistinct("value").alias("distinct_count"))
result.show()这段代码的功能是从每个行的数组中获取不同的计数。它首先将包含数组的DataFrame展开为多行，然后使用groupBy和countDistinct函数对每个行的数组元素进行计数。最后，打印出每个行的唯一计数结果。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark服务：https://cloud.tencent.com/product/spark
- 腾讯云数据仓库服务：https://cloud.tencent.com/product/dws
- 腾讯云数据计算服务：https://cloud.tencent.com/product/dc
- 腾讯云大数据服务：https://cloud.tencent.com/product/bds
相关·内容
3回答
电火花数据格式中所有列中的唯一元素数
python、apache-spark、dataframe、pyspark、apache-spark-sql
如何才能计算的每一列中的唯一元素数：
import pandas as pd
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = pd.DataFrame([[1, 100], [1, 200], [2, 300], [3, 100], [4, 100], [4, 300]], columns=['col1', 'col2'])
df_spark = spark.createDataFrame(df)
print(df_spark.show()
浏览 0提问于2018-12-13得票数 11
回答已采纳
2回答
如何从火花放电阵列中编码标签
python、apache-spark、pyspark、pyspark-sql
例如，我在DataFrame中有name中的分类特性
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local").appName("example")
.config("spark.some.config.option", "some-value").getOrCreate()
features = [(['a', 'b', 'c'], 1),
浏览 0提问于2018-12-04得票数 3
回答已采纳
1回答
如何自动将常量列放入火花放电中？
pyspark、apache-spark-sql
我有一个火花数据，我需要删除所有的常数列从我的数据。因为我不知道哪些列是常量，所以我不能手动取消选择常量列，也就是说，我需要一个自动过程。我很惊讶，我没有找到一个简单的解决方案的堆叠溢出。
示例：
import pandas as pd
import pyspark
from pyspark.sql.session import SparkSession
spark = SparkSession.builder.appName("test").getOrCreate()
d = {'col1': [1, 2, 3, 4, 5],
'col2
浏览 4提问于2019-04-21得票数 3
回答已采纳
1回答
列中的搜索值
python-3.x、pyspark、pyspark-dataframes
我想搜索列是否包含值。 import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pandas as pd
df_init = pd.DataFrame({'id':['1', '2'], 'val':[100, 200]})
spark = SparkSession.builder.appName('pandasToSparkDF').getOrCreate()
mySchema
浏览 16提问于2020-09-29得票数 1
回答已采纳
3回答
向PySpark数据帧中添加组计数列
apache-spark、pyspark、dplyr
我来自R和到PySpark，因为它的出色的火花处理，我正在努力从一个上下文映射到另一个特定的概念。
尤其是，假设我拥有如下数据集
x | y
--+--
a | 5
a | 8
a | 7
b | 1
我希望添加一个列，其中包含每个x值的行数，如下所示：
x | y | n
--+---+---
a | 5 | 3
a | 8 | 3
a | 7 | 3
b | 1 | 1
在dplyr中，我只想说：
import(tidyverse)
df <- read_csv("...")
df %>%
group_by(x) %>%
mutate(n
浏览 0提问于2018-02-14得票数 37
回答已采纳
2回答
根据包含列表的一列和其他作为值的列合并两个pyspark数据
python、pandas、dataframe、pyspark
我有两张桌子
+-----+-----+
|store|sales|
+-----+-----+
| F| 4000|
| M| 3000|
| A| 4000|
+-----+-----+`
+-----+------+
| upc| store|
+-----+------+
|40288|[F, M]|
|42114| [M]|
|39192|[F, A]|
+-----+------+`
我希望把最后一张桌子
+-----+------+-----+
| upc| store|sales|
+-----+------+-----+
|40288|[F, M]
浏览 6提问于2021-12-14得票数 0
回答已采纳
2回答
使用RDD列表作为数据过滤操作的参数
pyspark、spark-dataframe、rdd、pyspark-sql、apache-spark-2.0
我有下面的代码片段。
from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.types import *
sc = SparkContext()
spark = SparkSession.builder.appName("test").getOrCreate()
schema = StructType([
浏览 5提问于2017-09-15得票数 1
4回答
解析行并从Spark Dataframe中隔离学生记录
python-3.x、pyspark
我的学生数据库在表Student中有每个学生的多条记录。
我正在将数据读取到Spark数据帧中，然后迭代Spark数据帧，隔离每个学生的记录，并对每个学生记录进行一些处理。
到目前为止我的代码如下：
from pyspark.sql import SparkSession
spark_session = SparkSession \
.builder \
.appName("app") \
.config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:2.7.2"
浏览 48提问于2019-10-23得票数 0
1回答
正在读取Pyspark代码中的嵌套Json文件。pyspark.sql.utils.AnalysisException：
json、pyspark、explode
我正在尝试读取嵌套的JSON文件。我无法分解嵌套列并正确读取JSON文件。 My Json file
{
"Univerity": "JNTU",
"Department": {
"DepartmentID": "101",
"Student": {
"lastName": "Fraun",
"address": "23 hyd 500089&#
浏览 174提问于2021-07-09得票数 0
1回答
使用when语句对多个数据帧列进行协调
python-3.x、apache-spark、pyspark、apache-spark-sql、pyspark-dataframes
我正在尝试连接多个dataframe列，我无法在下面的concat_ws语句中执行pyspark或expr。
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
from pyspark.sql.functions import concat_ws,concat,when,col,expr
from pyspark.sql.functions import lit
from pyspark.sql import SparkSession
spark = SparkSession.build
浏览 1提问于2020-08-30得票数 0
回答已采纳
1回答
将重复值转换为两个不同的列
apache-spark、pyspark、apache-spark-sql、duplicates、pivot
我将以下数据导入到PySpark数据框架中：
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSession.builder.master("local[4]").appName("pyspark").getOrCreate()
df = spark.read.csv("example.csv")
df.show()
#+--+------+
#|id|ans
浏览 10提问于2022-05-05得票数 1
回答已采纳
2回答
给定以下DataFrame：
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local").appName("test").getOrCreate()
df = spark.createDataFrame([['a',1],['b', 2],['a', 3]], ['category', 'value'])
df.
浏览 2提问于2018-09-10得票数 7
回答已采纳
3回答
如何将DataFrame中的字符串列拆分为多个列？
pyspark
这是当前的代码：
from pyspark.sql import SparkSession
park_session = SparkSession\
.builder\
.appName("test")\
.getOrCreate()
lines = spark_session\
.readStream\
.format("socket")\
.option("host", "127.0.0.1")\
.option("port", 9998)\
浏览 0提问于2020-04-20得票数 3
回答已采纳
2回答
pyspark: rdd中包含列表列表的数据
list、dataframe、pyspark、rdd
我是斯派克(使用Python)的新手，甚至在浏览相关帖子之后，我也无法理解这一点。
我有个RDD。RDD的每个记录都是一个列表，如下所示
[[1073914607, 0, -1],[1073914607, 2, 7.88],[1073914607, 0, -1],[1073914607, 4, 40.0]]
[[1074079003, 0, -1],[1074079003, 2, 2.87],[1074079003, 0, -1],[1074079003, 4, 35.2]]
我想将RDD转换为3列的dataframe，基本上是堆叠所有元素列表。数据文件应该如下所示。
account_id
浏览 7提问于2021-07-16得票数 0
回答已采纳
2回答
PySpark DataFrame:查找最接近整数列值的数组列的索引
python、arrays、pyspark
我有一个PySpark DataFrame，它具有数组列类型和整数列类型。我想找出整数列在值方面最接近的数组位置。见下文：
df = spark.createDataFrame(
[
(1, [5, 20, 100, 250], 2),
(2, [16, 53, 120, 180], 168),
(3, [100, 200, 1000, 2500], 3500),
],
["id", "array_col", "int_col"]
)
我希望创建一个新列，该列查看int的值最接近的array_col中的哪个数组索引
浏览 2提问于2021-10-15得票数 0
回答已采纳
1回答
在星星之火中找到缺少的值后，转换为熊猫数据
pyspark、apache-spark-sql
我正在利用以下内容来查找星星之火中缺少的值：
from pyspark.sql.functions import col,sum
df.select(*(sum(col(c).isNull().cast("int")).alias(c) for c in df.columns)).show()
从我的样本火花df如下：
import numpy as np
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCr
浏览 4提问于2022-03-31得票数 1
回答已采纳
3回答
两个数组成员的乘法
arrays、apache-spark、pyspark、apache-spark-sql、multiplication
我有下表： from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.getOrCreate()
cols = [ 'a1', 'a2']
data = [([2, 3], [4, 5]),
([1, 3], [2, 4])]
df = spark.createDataFrame(data, cols)
df.show()
# +------+------+
# | a1| a2|
# +------+----
浏览 43提问于2021-06-28得票数 2
回答已采纳
2回答
如果组中存在非空项，如何删除重复项和空项？
dataframe、pyspark、filter、group-by
下面的DataFrame应该根据标志列进行过滤。如果基于列id和cod的组没有任何与None值不同的行，则只需要维护一个唯一的行，否则，必须删除列标志<code>E 211</code>中的None值行。
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number,max
spark = SparkSession.builder.appName(
浏览 3提问于2022-06-30得票数 0
1回答
火花放电中添加UUID的有效方法
python-3.x、apache-spark、pyspark
我有一个DataFrame，我想要添加一个由不同的uuid4()行组成的列。我的代码：
from pyspark.sql import SparkSession
from pyspark.sql import functions as f
from pyspark.sql.types import StringType
from uuid import uuid4
spark_session = SparkSession.builder.getOrCreate()
df = spark_session.createDataFrame([
[1, 1, 'teste
浏览 3提问于2020-03-11得票数 6
回答已采纳
3回答
计算spark数据帧中的字数
python、apache-spark、pyspark、apache-spark-sql
如果不使用SQL的REPLACE()函数，我们如何才能找到spark数据帧一列中的字数？下面是我正在使用的代码和输入，但replace()函数不起作用。
from pyspark.sql import SparkSession
my_spark = SparkSession \
.builder \
.appName("Python Spark SQL example") \
.enableHiveSupport() \
.getOrCreate()
parqFileName = 'gs://caserta-pyspark-eval/t
浏览 3提问于2018-02-22得票数 10
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
活动推荐
腾讯云开发者
