如何在不使用explode_outer的情况下扁平化PySpark 2.1.1模式来维护空数据点？
在PySpark 2.1.1中，如果不想使用explode_outer函数来扁平化模式并维护空数据点，可以使用select和alias函数来实现。
首先，使用select函数选择需要扁平化的列，并使用alias函数为每个列创建一个别名。然后，使用coalesce函数将空数据点替换为一个特定的值。最后，使用select函数选择所有别名列，并使用alias函数为整个结果创建一个别名。
以下是一个示例代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, coalesce
# 创建SparkSession
spark = SparkSession.builder.getOrCreate()
# 创建示例数据
data = [(1, "John", 25, None), (2, "Alice", None, "New York"), (3, "Bob", 30, "London")]
df = spark.createDataFrame(data, ["id", "name", "age", "city"])
# 扁平化模式并维护空数据点
flattened_df = df.select(
col("id").alias("id"),
col("name").alias("name"),
coalesce(col("age"), col("age").cast("string")).alias("age"),
coalesce(col("city"), col("city").cast("string")).alias("city")
).select(
col("id"),
col("name"),
col("age"),
col("city")
).alias("result")
# 显示结果
flattened_df.show()在上述示例中，我们选择了需要扁平化的列，并为每个列创建了一个别名。然后，使用coalesce函数将空数据点替换为一个特定的值，这里我们将空数据点替换为列本身的字符串表示。最后，我们选择所有别名列，并为整个结果创建了一个别名。
这样，我们就可以在不使用explode_outer的情况下扁平化PySpark 2.1.1模式并维护空数据点。
相关·内容
3回答
我有两个要比较的数据帧，它们都有相同的列数，比较结果应该有不匹配的字段和ID值。
数据一号
+-----+---+--------+
| name| id| City|
+-----+---+--------+
| Sam| 3| Toronto|
| BALU| 11| YYY|
|CLAIR| 7|Montreal|
|HELEN| 10| London|
|HELEN| 16| Ottawa|
+-----+---+--------+
数据二
+-------------+-----------+-------------+
|Expected_name|Expe
浏览 2提问于2018-08-16得票数 7
回答已采纳
我正在处理PySpark数据帧中的一些深度嵌套数据。当我试图将结构扁平化为行和列时，我注意到当我调用withColumn时，如果该行在源列中包含null，那么该行将从我的结果数据帧中删除。相反，我希望找到一种方法来保留行，并在结果列中使用null。
要使用的示例数据帧：
from pyspark.sql.functions import explode, first, col, monotonically_increasing_id
from pyspark.sql import Row
df = spark.createDataFrame([
Row(dataCells=[Row(po
浏览 0提问于2018-10-11得票数 2
1回答
我试图修改Map类型的Dataframe列，其值依次为Struct类型：
root
|-- name: string (nullable = true)
|-- age: long (nullable = true)
|-- MapName: map (nullable = true)
| |-- key: string
| |-- value: struct (valueContainsNull = true)
| | |-- field1: double (nullable = true)
| | |-- field2: double (n
浏览 3提问于2021-01-04得票数 3
回答已采纳
我正在尝试比较两个数据帧，以查找新记录和更新的记录，这些记录将用于创建第三个数据帧。我使用的是Pyspark 2.4.3
由于我来自SQL背景(ASE)，我最初的想法是做一个左连接来查找新记录，并对所有列的散列进行!=操作来查找更新：
SELECT a.*
FROM Todays_Data a
Left Join Yesterdays_PK_And_Hash b on a.pk = b.pk
WHERE (b.pk IS NULL) --finds new records
OR (b.hashOfColumns != HASHBYTES('md5',<converted
浏览 0提问于2019-07-12得票数 0
1回答
我目前正在尝试对服务列进行一些聚合。我想将所有相似的服务分组并求和，如果可能的话，将其展平到一行中。
输入：
+------------------+--------------------+
| cid | Services|
+------------------+--------------------+
|845124826013182686| [112931, serv1]|
|845124826013182686| [146936, serv1]|
|845124826013182686| [32718, s
浏览 23提问于2018-06-15得票数 0
回答已采纳
1回答
代码版本：
Python==3.7Spark Version==2.4.7Pyspark==2.4.5Hive==2.3.7
你好，希望有人能帮我这个忙。我正在使用PySpark读取几个大型文件(每个文件大约80 GB，其中6个左右)。
使用一个领导节点，两个工作节点。
正在读取的文件有1000多个cols，数百万行上有数百万。当我试图处理这个文件时，它会以不一致的列数推送列值。
我尝试过的事情：
查看了每行的分隔符数。这个数字在各个行之间似乎是一致的，但是我只看过一个文件，通过使用read.text(s3_path)和对每个列使用substr来尝试固定宽度。这是行不通的，试图从空字符串中推断
浏览 2提问于2021-02-16得票数 1
1回答
我正在开发基于一组ORC文件的spark数据访问sql查询。这个程序是这样的：
from pyspark.sql import SparkSession
spark_session = SparkSession.builder.appName("test").getOrCreate()
sdf = spark_session.read.orc("../data/")
sdf.createOrReplaceTempView("test")
现在我有了一张名为“测试”的桌子。如果我做这样的事：
spark_session.sql("selec
浏览 0提问于2018-02-16得票数 5
回答已采纳
1回答
我正在使用一个自定义函数作为reduce操作的一部分。对于下面的例子，我得到了下面的消息TypeError: reduce() takes no keyword arguments -我相信这是由于我在函数exposed_colum中使用字典mapping的方式-你能帮我修复这个函数吗？
from pyspark.sql import DataFrame, Row
from pyspark.sql.functions import col
from pyspark.sql import SparkSession
from functools import reduce
def proces
浏览 18提问于2020-02-05得票数 0
回答已采纳
1回答
我有下面的data.frame ( spark )
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
from pyspark.sql import functions as sf
from pyspark.sql.functions import col, when, lit
ddf = spark.createDataFrame([[None, 'Mic
浏览 2提问于2018-05-28得票数 2
回答已采纳
我现在遇到了一些问题。
我有一大堆股票滴答数据csv文件。我需要计算每只股票的秒级数据。
我希望每个股票的数据可以在10秒内处理，合并成一个大文件，最后输出到csv。因为使用熊猫会受到我笔记本电脑记忆的限制，如果我想在熊猫身上做的话，我需要做很多读csv/ to _csv的工作。我认为这将占用时间，所以我选择了这样一种方式：(1)我使用pyspark读取所有的csv文件，生成一个大文件df。(2)我从df中获得了股票列表。然后进行迭代，每次我选择一个股票数据的pyspark，把它转移到熊猫数据中，在熊猫中计算它。最后，将该文件输出到本地文件中。
现在的问题是程序运行速度非常慢，对于一些股票，它
浏览 4提问于2020-04-19得票数 0
3回答
我来自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
回答已采纳
我有一个带有n个cols的percentage ( df )，我想要生成另一个n cols的df，其中每一列都记录相应的原始df列中的b/w连续行的百分比差。新df中的列标题应该是旧dataframe + "_diff“中相应的列标题。使用下面的代码，我可以为原始df中的每一列生成百分比变化的新列，但无法将它们插入具有适当列标题的新df中：
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
import pyspark.sql.functions as func
spark = (Sp
浏览 0提问于2018-05-31得票数 0
回答已采纳
有一个列和表名列表的表。
我必须编写一个查询来列出一个SELECT查询。
示例:这是我从JSON中提取的信息。现在我必须构建一个SQL语句。
ROW_ID predicateName EntityName
1 AGE tbl_PR_AGE_VALIDATION_RULE
2 SALARY tbl_PR_AGE_VALIDATION_RULE
3 GENDER tbl_PR_AGE_VALIDATION_RULE
4 DATEOBIRTH tbl_PR_AGE_VALIDATION
浏览 0提问于2019-11-14得票数 3
回答已采纳
1回答
我正在研究 stuck 方法，并坚持了如何实现以下结果。
星星之火sql (这很有效)
q = """
select breed,
avg(weight) as avg_wt,
avg(weight) filter (where age > 1) avg_wt_age_gt1
from cats
group by breed
order by breed
"""
spark.sql(q).show()
问:如何使用方法获得相同的结果？
我的尝试
(sdf.groupBy("breed").
浏览 2提问于2020-07-23得票数 0
回答已采纳
假设我有两个MySQL表：
table1 (模板)列: id_template，name，col3，col4，.
table2 (别名)列: id_alias、id_template_orig、id_user、别名
我想从table1 (模板)中选择一行，包括所有的列(*)。在相同的select语句中，我想检查当前用户是否在table2 (别名)中为原始模板名保存了别名。将别名列包含在带有左联接的结果行中没有问题，但我想知道是否有一种方法可以用table1别名值替换已经从table2中选择的原始name列值。
下面是到目前为止的选择：
SELECT table1.*, table2.alias
浏览 2提问于2020-11-08得票数 0
回答已采纳
假设我在pySpark中有以下数据帧：
df1 = sqlContext.createDataFrame([Row(name='john', age=50), Row(name='james', age=25)])
df2 = sqlContext.createDataFrame([Row(name='john', weight=150), Row(name='mike', weight=115)])
df3 = sqlContext.createDataFrame([Row(name='john', age=50
浏览 0提问于2015-08-23得票数 7
回答已采纳
我正在使用Spark3.1.1版本的PySpark DataFrame API进行本地设置。在读取数据、执行一些转换等之后，我将DataFrame保存到磁盘。输出目录与part-0000*文件一起创建，输出目录中也有_SUCCESS文件。但是，我的part-0000*总是空的，即零字节。
我试着用parquet和csv格式编写它，结果是一样的。在编写之前，我调用了df.show()以确保DataFrame中有数据。
### code.py ###
from pyspark.sql import SparkSession
from pyspark.sql import functions as
浏览 8提问于2022-05-11得票数 1
2回答
重复循环
、、、
我有两个数据帧，我正在检查dataframe A中的数据，如果列为null，那么在dataframe B中用相同的列替换空数据。
两个数据格式都有唯一的ID列，根据我正在加入的数据格式，下面的代码都很好。
updated_data = TABLE_BY_updated_date_unique.select('name_id_forwarded','name_id','name_id_org','first','last','passport','PHONE','EMAIL&#
浏览 5提问于2020-08-09得票数 4
回答已采纳
1回答
每当我加载文件并打印模式时，它都会显示一堆标记为col4-col89的空列(您的数字可能会有所不同)，而当我写入csv时，它会显示额外的空列。
我的代码示例：
%pyspark
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.
浏览 7提问于2019-10-29得票数 0
然而，我想要完成的正是这个问题()；在我的例子中，我使用的是Python/Pyspark，而不是Scala。
我正在尝试提取Kafka连接消息的“有效载荷”部分，其中也包括模式。
示例消息：
{"schema":{"type":"struct","name":"emp_table","fields":[{"field":"emp_id","type":"string"},{"field":"emp_name
浏览 0提问于2020-10-11得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
