Pyspark:如何转换dataframe列中的json字符串
Pyspark是一种基于Python的Spark编程接口，用于处理大规模数据集的分布式计算。在Pyspark中，要转换dataframe列中的json字符串，可以使用Spark的内置函数和方法来实现。
首先，需要导入相关的模块和函数：
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, StringType接下来，创建一个SparkSession对象：
spark = SparkSession.builder.appName("JsonTransformation").getOrCreate()然后，定义一个示例的dataframe：
data = [
('{"name":"John", "age":30, "city":"New York"}'),
('{"name":"Alice", "age":25, "city":"San Francisco"}'),
('{"name":"Bob", "age":35, "city":"Los Angeles"}')
]
df = spark.createDataFrame(data, StringType()).toDF("json_string")定义一个schema，用于解析json字符串：
schema = StructType([
StructField("name", StringType(), True),
StructField("age", StringType(), True),
StructField("city", StringType(), True)
])使用from_json函数将json字符串转换为结构化的列：
df = df.withColumn("json_struct", from_json(df.json_string, schema))最后，可以选择性地提取json中的字段：
df = df.select("json_struct.name", "json_struct.age", "json_struct.city")这样，就完成了dataframe列中json字符串的转换。你可以根据实际需求选择需要的字段进行进一步处理或分析。
腾讯云相关产品推荐：腾讯云的云数据库TDSQL，可以提供高性能、高可用、可扩展的数据库服务，适用于各种应用场景。你可以通过以下链接了解更多信息：腾讯云数据库TDSQL
注意：以上答案仅供参考，具体的解决方案可能因实际情况而异。
相关·内容
1回答
当我尝试通过show()和display( dataframe )显示来自pyspark dataframe的date列时，这些数据列的格式是不同的。现在，我们如何得出数据帧中存在哪种日期格式？ 显示: 2018-02-15T06:47:19.000+0000 演出时间: 2018-02-15 06:47:19 ?
浏览 15提问于2021-04-23得票数 2
我正在使用pyspark中的to_json将dataframe中的结构列转换为json列，但是在json中忽略了少数结构字段中的空值，我不希望这些空值被忽略。
浏览 6提问于2020-10-14得票数 0
1回答
我有一个使用结构化星火流python进行实时处理的任务，所以第一步是将csv文件吞入kafka主题:完成。
第二步是卡夫卡主题的readStream。
df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
.option("subscribe", kafka_topic_name) \
.option("startingOffsets", &#
浏览 2提问于2021-09-25得票数 1
回答已采纳
我有一个从熊猫数据帧创建的样本spark数据帧-
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import StringType
from pyspark.sql.types import *
import pandas as pd
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config(&
浏览 3提问于2020-09-09得票数 0
我试图在pyspark中创建空的dataframe，在pyspark中，我从外部JSON文件传递scehma，但是Json不允许我指定struct类型，所以我提到它是string。json文件： "OptionalEvents" : {
"Event1":
"StructType([StructField('id',StringType(), True),StructField('time',StringType(), True),StructField('ts',StringTyp
浏览 39提问于2021-08-23得票数 1
回答已采纳
我从Kafka那里得到了JSON格式的数据，并在PySpark中以DataFrame的形式读取了这些数据。 在我从Kafka获得数据后，它显示为DataFrame格式： DataFrame[value: string] 但是，该值包含JSON / DICT格式。 打印语句和返回： def print_row(row):
print(row)
pass
testing.writeStream.foreach(print_row).start() Row(value='{col_1 =80.0, timestamp=2020-01-13T08:58:58.164Z}&#
浏览 24提问于2020-01-13得票数 0
我有一个PySpark DataFrame df，它有一些列，如下所示。hour列采用UTC时间，我想创建一个基于time_zone列的具有本地时间的新列。我如何在PySpark中做到这一点？
df
+-------------------------+------------+
| hour | time_zone |
+-------------------------+------------+
|2019-10-16T20:00:00+0000 | US/Eastern |
|2019-10-15T23:0
浏览 2提问于2019-12-03得票数 5
回答已采纳
1回答
如何将Vectors.dense列添加到pyspark中？
import pandas as pd
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.ml.linalg import DenseVector
py_df = pd.DataFrame.from_dict({"time": [59., 115., 156., 421.], "event": [1, 1, 1, 0]})
sc = SparkContext(master="loc
浏览 1提问于2018-04-14得票数 1
回答已采纳
我有一个CSV文件，它包含JSON对象以及其他数据，比如String，Integer。如果我尝试将文件读取为CSV，那么JSON对象将在其他列中重叠。
Column1, Column2, Column3, Column4, Column5
100,ABC,{"abc": [{"xyz": 0, "mno": "h"}, {"apple": 0, "hello": 1, "temp": "cnot"}]},foo, pine
101,XYZ,{"xyz&#
浏览 2提问于2020-09-23得票数 1
以下pyspark命令 df = dataFrame.groupBy("URL_short").count().select("URL_short", col("count").alias("NumOfReqs")) 创建了以下结果。 |URL_short |NumOfReqs|
+-----------------------------------------------------------------------------------------+---------+
|http1 | 500 |
|h
浏览 19提问于2018-12-18得票数 1
回答已采纳
首先，我要预先感谢大家的帮助！我有4张桌子，我加入了它们，得到了一个PySpark数据。其中一个dataframe列如下所示，它有大约20万条记录：
{"table_name":"BTR.DAILY_BTR.JSC_MON","login":"0015471"}
{"table_name":"BTR.DAILY_BTR.ESHOP.JSC_MON","login":"0015471"}
该列的类型为“string”。我需要通过key table_name获得价值。我尝
浏览 4提问于2022-04-26得票数 0
我想使用PySpark (Spark1.6.2)对存在于Hive表中的数值数据执行主成分分析(PCA)。我能够将Hive表导入到：
>>> from pyspark.sql import HiveContext
>>> hiveContext = HiveContext(sc)
>>> dataframe = hiveContext.sql("SELECT * FROM my_table")
>>> type(dataframe)
<class 'pyspark.sql.dataframe.D
浏览 1提问于2016-10-06得票数 4
回答已采纳
1回答
我使用的是PySpark版本3.0.1。我正在将csv文件读取为具有2个日期列的PySpark数据帧。但是，当我尝试打印模式时，两列都被填充为字符串类型。 ? ? 上面附加的屏幕截图是Dataframe和Dataframe模式。 如何使用pyspark将date列中的行值转换为时间戳格式？ 我已经尝试了很多东西，但所有的代码都需要当前的格式，但如何转换为正确的时间戳，如果我不知道什么格式的csv文件。 我也尝试过下面的代码，但这是在创建一个空值的新列 df1 = df.withColumn('datetime', col('joining_date&#
浏览 16提问于2020-12-30得票数 2
我有一个用例来映射基于条件的pyspark列的元素。通过这个文档，我找不到一个函数来执行映射函数。
因此，尝试使用pyspark映射函数，但无法将pyspark列转换为dataFrame
注意:我之所以使用pyspark列，是因为我从我使用的库(远大期望)中获得了它的输入。
@column_condition_partial(engine=SparkDFExecutionEngine)
def _spark(cls, column, ts_formats, **kwargs):
return column.isin([3])
# need to replace the abov
浏览 4提问于2021-11-19得票数 0
回答已采纳
我正在使用pyspark读取下面的json文件：
{
"data": {
"indicatr": {
"indicatr": {
"id": "5c9e41e4884db700desdaad8"}}}}
我写了下面的python代码：
from pyspark.sql import Window, DataFrame
from pyspark.sql.types import *
from pyspark.sql.types import StructType
from py
浏览 22提问于2019-11-28得票数 0
我正在尝试将前导零添加到pyspark dataframe中的一列 输入:- ID 123 预期输出： 000000000123
浏览 17提问于2019-09-16得票数 5
回答已采纳
在pyspark中，有没有办法将时间戳数据类型的dataframe列转换为格式为'YYYY-MM-DD‘格式的字符串？
浏览 3提问于2018-02-22得票数 14
回答已采纳
1回答
如何在pyspark中处理dataframe并获得json格式的输出：
数据帧：
empid empname in out
1 A 1 1
1 A 1 1
json中需要的输出：
{
id:empid,
name:empname,
in:[1,1],
out:[1,1]
}
浏览 4提问于2018-03-18得票数 0
1回答
我有一个PySpark数据帧，它有一个复杂的列，请参考下列值：
ID value
1 [{"label":"animal","value":"cat"},{"label":null,"value":"George"}]
我想在PySpark dataframe中添加一个新列，它基本上将它转换为一个字符串列表。如果Label为null，则字符串应包含value；如果label不为null，则string应为label:value。因此，对于上面的示例数据帧，输出应如下所示：
ID
浏览 0提问于2021-02-09得票数 0
我正在将一个dataframe转换成一个管道分隔的值，并将其写入shell (scala)中的一个文件中。但我在PySpark方面毫无头绪。会很感激你的帮助。
尤其是我不知道如何用“\”来连接每个列
这是scala版本
scala> val stgDF = spark.read.table("tbl")
stgDF: org.apache.spark.sql.DataFrame = [name: string, num: int]
scala> stgDF.map(line => line.mkString("|")).take(2) //H
浏览 3提问于2017-10-30得票数 4
回答已采纳
2回答
我有一个使用以下命令创建的pyspark.sql.dataframe.DataFrame实例
dataframe = sqlContext.sql("select * from table").
其中一列是“arrival_date”，其中包含一个字符串。
如何修改此列，使其只取其中的前4个字符，并丢弃其余的字符？
如何将此列的类型从字符串转换为日期？
在graphlab.SFrame中，这将是：
dataframe['column_name'] = dataframe['column_name'].apply(lambda x: x[:4] )
浏览 0提问于2016-08-20得票数 4
回答已采纳
有没有办法用pyspark 2.1.0创建/填充列，其中列的名称是另一个列的值？我尝试了以下几种方法
def createNewColumnsFromValues(dataFrame, colName, targetColName):
"""
Set value of column colName to targetColName's value
"""
cols = dataFrame.columns
#df = dataFrame.withColumn(f.col(colName), f.col(targetCol
浏览 0提问于2018-09-13得票数 1
我正在尝试将spark数据框中的列拆分为多个值。因此，我分隔了要拆分的列，并检查了其类型：
I: type(TopicModelVectorSplit)
O: pyspark.sql.dataframe.DataFrame
当我使用以下命令分割字符串时：
TopicModelVectorSplit = TopicModelVector.select('TopicModelVector').map(lambda line: line.text.split())
它会转换成流水线的RDD，而这些是不可能使用的。
I: type(TopicModelVectorSplit)
O: p
浏览 0提问于2016-06-22得票数 0
我正在使用PySpark，并且需要将DataFrame中的每一行转换为JSON文件(在s3中)，最好使用选定列的值来命名该文件。
我不知道该怎么做。任何帮助都将不胜感激。
浏览 10提问于2019-08-05得票数 0
1回答
我刚刚找到了一种将我的String数据run中的所有PySpark类型变量转换为分类变量的方法，这样我就可以在dataframe上运行决策树了。由于资源的限制，我不能使用熊猫，只能使用PySpark库。我已经确定了VectorIndexer是一种可能的解决方案，但是，我不明白如何转换所有String类型列，而认为这些列都是可能的。
谁能帮我弄清楚怎么做的语法吗？我想要的是这样的东西：
featureIndexer = VectorIndexer(inputCol=<list of input columns>, outputCol=<list of output column
浏览 1提问于2017-10-10得票数 0
回答已采纳
我有一个pyspark dataframe，它有一个值为string json的列。如何计算与字典内列表中的某个值匹配的值，并以列的形式返回报告？我想使用Python函数和pyspark udf来实现。 例如，下面是数据帧df： +---------------------------------------------------------------------------+
|col |
+---------------------------
浏览 15提问于2020-12-11得票数 0
4回答
我有一个具有下面数据(所有列都有字符串数据类型)的Pyspark (原始Dataframe)。在我的用例中，我不确定输入数据中有哪些列。用户只需将dataframe的名称传递给我，并要求我修剪这个dataframe的所有列。典型数据文件中的数据如下所示：
id Value Value1
1 "Text " "Avb"
2 1504 " Test"
3 1
浏览 2提问于2017-07-31得票数 0
我有一个dataframe，它在一个列中存储一个JSON对象。我希望处理JSON对象来创建一个新的dataframe (列的数目和类型不同，每行将从JSON对象生成n个新行)。我在下面编写了这样的逻辑:在迭代原始数据集时，将字典(行)附加到列表中。
data = []
def process_row_data(row):
global data
for item in row.json_object['obj']:
# create a dictionary to represent each row of a new dataframe
浏览 4提问于2022-10-19得票数 1
回答已采纳
1回答
我目前正在处理一些相当复杂的json文件，我应该将它们转换并写入增量表。问题是，当涉及到列的数据类型时，每个文件都有细微的差异。有人能给我解释一下检索嵌套结构列的数据类型的一般方法吗？在互联网上，我只能找到如何对它们做选择：https://sparkbyexamples.com/pyspark/pyspark-select-nested-struct-columns/ 如果我有这样的格式： ? 我怎样才能获得数据类型，比如说，姓氏？ 编辑: Json文件当然已经写在dataframe中，我的问题是如何查询dataframe以检索数据类型 非常感谢!
浏览 42提问于2021-10-22得票数 0
我尝试创建并使用日期值填充pyspark dataframe。 Columns = ["EmployeeNo", "Name", "EmployeeID", "ValidFrom", "ValidTo"]
Data = [(100, "Hilmar Buchta", "HB", "2000-01-01", "2999-12-31"),
]
DfEmployee = spark.createDataFrame(Data, Columns
浏览 41提问于2021-02-23得票数 1
回答已采纳
2回答
我正在尝试解析单个列的columns.My数据帧，并获取具有多个pyspark数据帧的数据帧，如下所示：
a b dic
0 1 2 {'d': 1, 'e': 2}
1 3 4 {'d': 7, 'e': 0}
2 5 6 {'d': 5, 'e': 4}
我想要解析dic列并获得数据帧，如下所示。如果可能的话，我期待着使用pandas UDF。我的预期输出如下：
a b c d
0 1 2 1 2
1 3 4 7
浏览 40提问于2020-03-01得票数 0
回答已采纳
1回答
使用模式读取固定宽度文件
、、、
我有固定宽度的文件，如下所示
00120181120xyz12341
00220180203abc56792
00320181203pqr25483
以及相应的JSON文件，该文件指定架构：
{"Column":"id","From":"1","To":"3"}
{"Column":"date","From":"4","To":"8"}
{"Column":"name",
浏览 0提问于2018-12-17得票数 3
回答已采纳
5回答
查看新的spark API，还不清楚是否有可能修改DataFrame列。
如何更改数据文件的行x列y中的值？
在pandas中，这将是：
df.ix[x,y] = new_value
编辑：合并下面所说的内容，您不能修改现有的数据格式，因为它是不可变的，但是您可以返回一个新的数据格式，并进行所需的修改。
如果只想根据条件替换列中的值，如np.where
from pyspark.sql import functions as F
update_func = (F.when(F.col('update_col') == replace_val, new_value)
浏览 12提问于2015-03-17得票数 95
回答已采纳
我是在Databricks中使用Pyspark的新手，所以这就是为什么我要与以下内容斗争的原因:我有一个dataframe，它有数据类型为string的列。一个例子(见下图)是列netto_resultaat。 要求将此列和其他与财务相关的列从字符串更改为小数。这是因为我要导出Azure SQL数据库中的架构和数据。 我尝试了以下几种方法： 从pyspark.sql.functions导入列从pyspark.sql.types导入DecimalType newouterjoindffinal = newouterjoindf.withColumn("netto_resultaat&#
浏览 23提问于2021-10-25得票数 0
回答已采纳
1回答
在PySpark中将csv文件转换为parquet时遇到问题。当转换相同模式的多个文件时，它们不具有相同的模式，因为有时数字字符串将被读取为浮点型，其他字符串将被读取为整数，等等。列的顺序似乎也有问题。似乎当编写具有相同列的数据帧时，但是以不同的顺序排列以拼接，那么这些拼接就不能被加载到相同的语句中。 如何将dataframe写入parquet，以使所有列都以字符串类型存储？如何处理列的顺序？在写入拼图之前，我是否应该将所有数据帧的列重新排列为相同的顺序？
浏览 23提问于2021-01-20得票数 0
如果我有一个有两个列的PySpark DataFrame，text和subtext，其中subtext肯定会出现在text的某个地方。如何计算subtext在text列中的位置？
输入数据：
+---------------------------+---------+
| text | subtext |
+---------------------------+---------+
| Where is my string? | is |
| Hm, this one is different | on |
+--
浏览 3提问于2021-01-21得票数 2
回答已采纳
3回答
我有以下数据-
{
"Id": "01d3050e",
"Properties": "{\"choices\":null,\"object\":\"demo\",\"database\":\"pg\",\"timestamp\":\"1581534117303\"}",
"LastUpdated": 1581530000000,
"LastUpdatedBy"
浏览 2提问于2020-02-19得票数 2
给出一个具有非空uuid列和可空uuid列的表设计，如何使用Python3.7.9与Pysmack2.4.3数据table和PostgreSQL42.2.18.jar驱动程序进行插入？
table_df = spark.read.format('jdbc) \
.option('driver', 'org.postgresql.Driver') \
.option('dbtable', 'example_table') \
浏览 9提问于2020-11-03得票数 4
我正试图通过启用inferSchema来将csv文件读入火花df，但随后无法获得fv_df.columns。下面是错误消息
>>> fv_df = spark.read.option("header", "true").option("delimiter", "\t").csv('/home/h212957/FacilityView/datapoints_FV.csv', inferSchema=True)
>>> fv_df.columns
Traceback (most re
浏览 12提问于2017-04-26得票数 0
我见过很多类似的问题被问了很多次，但是对于一些应该很容易的问题，没有明确的答案。
如何在PySpark中将结构列保存到CSV (tsv实际)？我想序列化它并将其保存为JSON。
我有一个dataframe，它包含以下模式，我从拼花中读到它：
timestamp:long
timezoneOffset:string
dayInterval:integer
speed:double
heading:double
ignitionStatus:integer
segmentId:string
pointMM:struct
mmResult:array
element:stru
浏览 2提问于2022-09-20得票数 1
回答已采纳
1回答
在下面的代码中，数据文件的所有列都是字符串。其中一列用一个小数位存储整数或小数(6.1,4.8,3,9.4,6，...etc.)。但是，一旦将数据加载到pyspark dataframe中，它也会显示带有单个小数位(例如3.0)的整数。
问题：我们如何才能强迫pyspark显示所有不带小数的整数值？例如，3.0应该显示为3。
from pyspark.sql.types import StringType
from pyspark.sql import functions as F
df = spark.read.csv(".......dfs.core.windows.net/my
浏览 7提问于2022-05-21得票数 0
1回答
我在Apache，Databricks上用PySpark编写代码。
我有一个DataFrame DF，DataFrame包含以下列A、B、C、D、E、F、G、H、I、J。
以下内容验证了dataframe是否具有所需的列
has_columns(very_large_dataframe, ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J'])
要求将2.5的换算系数适用于F栏，即值2，换算
浏览 6提问于2022-01-14得票数 0
回答已采纳
我想使用PySpark的RegexTokenizer从DataFrame列中删除5位或更多位的数字。我可以使用下面的代码提取这些数字，但有人知道我如何删除它们吗？
我的代码：
regexTokenizer = RegexTokenizer(inputCol="description", outputCol="tokenized_description", gaps=False,pattern="[0-9]{5,}")
如果字符串是"123abc 122323232"，我希望它去掉122323232，变成"123abc“
浏览 0提问于2019-02-13得票数 0
我正试图在我的数据集上运行PySpark中的PySpark算法。
from pyspark.ml.fpm import FPGrowth
fpGrowth = FPGrowth(itemsCol="name", minSupport=0.5,minConfidence=0.6)
model = fpGrowth.fit(df)
我得到了以下错误：
An error occurred while calling o2139.fit.
: java.lang.IllegalArgumentException: requirement failed: The input
col
浏览 0提问于2018-04-05得票数 2
回答已采纳
Scala有一个Dataset#transform方法，可以轻松地链接自定义的DataFrame转换，如下所示：
val weirdDf = df
.transform(myFirstCustomTransformation)
.transform(anotherCustomTransformation)
我没有看到与transform方法等价的方法。
是否有一种链接自定义转换的PySpark方法？
如果不是，如何修补pyspark.sql.DataFrame类以添加transform方法？
更新
变换方法为 ( )。
浏览 0提问于2017-09-15得票数 7
回答已采纳
1回答
我有一个嵌套的JSON字典，我需要将其转换为spark dataframe。此JSON字典显示在数据框列中。我一直在尝试使用"from_json“和"get_json_object”来解析dataframe列中存在的字典，但一直无法读取数据。以下是我一直在尝试读取的源数据的最小片段： {"value": "\u0000\u0000\u0000\u0000/{\"context\":\"data\"}"} 我需要提取嵌套的dict值。我使用下面的代码清理数据并将其读取到数据帧中 from pyspark.sql.f
浏览 36提问于2021-04-01得票数 1
回答已采纳
1回答
在PySpark中，如何将Dataframe转换为普通字符串？
背景：
我将PySpark与Kafka一起使用，而不是硬编码代理名称，而是在PySpark中参数化了Kafka broker名称。
Json文件保存了代理详细信息，Spark读取这个Json输入并将值赋值给变量。这些变量为带有字符串的Dataframe类型。
当我将dataframe传递给Pyspark连接细节以替换值时，我将面临问题。
错误：
只能将字符串(不是Dataframe)连接到字符串。
Json参数文件：
{
"broker": "https://at.com:8082",
"t
浏览 0提问于2021-03-05得票数 0
我在PySpark (Databricks)中有以下数据。如何将显示在dataframe中的确切文字时间戳提取为字符串？现在，使用下面的代码，我得到了："2022-02-25 06:32:29"而不是一些转换正在发生，其中包括“T”在内的毫秒部分。我想保留显示在dataframe上的文字字符串吗？
码
table = [x["ts"] for x in ts.rdd.collect()]
for row in table:
print(row)
浏览 2提问于2022-02-25得票数 1
1回答
我有这样的要求：
在Databricks中，我们正在读取csv文件。该文件有多个列，如emp_name、emp_salary、joining_date等。当我们在dataframe中读取该文件时，我们将所有的列作为字符串。
我们有一个API，它将给我们列的模式。emp_name是字符串(50)，emp_salary是十进制(7，4)，joining_date是时间戳等等。
我必须使用来自API的模式创建一个parquet文件。
我们如何在使用PySpark的Databricks中做到这一点。
浏览 4提问于2022-07-30得票数 -1
回答已采纳
我有一个用pyspark写的代码。我需要将其转换为字符串，然后将其转换为日期类型，等等。
我找不到任何方法来将此类型转换为字符串。我尝试了str()和.to_string()，但都不起作用。我把代码放在下面。
from pyspark.sql import functions as F
df = in_df.select('COL1')
> type(df)
> <class 'pyspark.sql.dataframe.DataFrame'>
> df.printSchema()
> |-- COL1: offset
浏览 7提问于2019-07-07得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
