用于数组的pyspark输入读取模式和用于读取json的结构
对于数组的pyspark输入读取模式，可以使用wholeTextFiles方法来读取。该方法可以读取一个目录下的所有文件，并返回一个键值对RDD，其中键是文件路径，值是文件内容。读取数组时，可以将每个文件的内容作为一个数组元素。
示例代码如下：
from pyspark import SparkContext
sc = SparkContext()
# 读取数组
array_rdd = sc.wholeTextFiles("path/to/directory")
# 打印数组内容
for file_path, content in array_rdd.collect():
print("File:", file_path)
print("Content:", content)对于读取JSON的结构，可以使用spark.read.json方法来读取JSON文件。该方法会将JSON文件解析为DataFrame，可以方便地进行数据处理和分析。
示例代码如下：
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
# 读取JSON文件
json_df = spark.read.json("path/to/json/file")
# 显示DataFrame内容
json_df.show()以上是使用pyspark进行数组输入读取和JSON结构读取的基本方法。在实际应用中，可以根据具体需求选择不同的读取模式和方法。
相关·内容
当我执行ValueError或df.columns时，我有一个火花数据文件，因为有一个数据类型为十进制(6，-12)，所以我得到错误的数据类型: decimal(6，-12)。
df = spark.read.csv("data.csv",inferSchema=True,header=True)
df.columns
运行df.columns或df.dtypes会出现以下错误
---------------------------------------------------------------------------
ValueEr
浏览 3提问于2022-01-19得票数 2
1回答
问题
最近，我在中遇到了一个挑战，当时我试图读取一个大型UTF-8JSONArray文件，并切换到HDInsight PySpark (v2.x，而不是3)来处理该文件。该文件为~110 g，具有~150 Objects对象。
HDInsight PySpark似乎不支持JSON文件格式的数组输入，所以我被困住了。而且，我有“许多”这样的文件，每个文件中都包含不同的模式，每个列都包含100列，因此现在不能为这些列创建模式。
问题
如何在PySpark 2中使用HDInsight上的开箱即用功能来使这些文件被读取为JSON？
谢谢,
J
Things我试过
我使用了这个页面底部的方法：，它提供了
浏览 1提问于2018-02-10得票数 0
回答已采纳
2回答
例如，给定以下json (命名为‘json’)：
{"myTime": "2016-10-26 18:19:15"}
和下面的python脚本：
from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext
conf = SparkConf().setAppName('simpleTest')
sc = SparkContext(conf=conf)
sqlContext = SQLContext(s
浏览 8提问于2017-05-09得票数 0
回答已采纳
1回答
我试着用这个pyspark代码读取一个csv文件：
tr_df = spark.read.csv("/data/file.csv",
header=True, inferSchema=True
)
tr_df.head(5)
但我知道这个错误：
~/anaconda3/envs/naboo-env/lib/python3.6/site-packages/pyspark/sql/dataframe.py中的 ValueError跟踪(最近一次调用)->1 tr_df.head(5)
在
浏览 3提问于2019-09-26得票数 0
回答已采纳
我有许多结构混乱的JSON文件。我想把这些解析成一个PySpark DataFrame。我编写了一个解析函数，并希望将其应用于目录中的每个文件，并将其输入到我的DataFrame中。
我对PySpark完全陌生，所以任何帮助都是非常感谢的。我曾希望我能做一些事情来达到这样的效果：
from my_parsers import parser
data_files = [...] # Is there a PySpark way to iterate through files in a directory?
my_data = map(parser, data_files) # How do
浏览 4提问于2022-02-28得票数 0
回答已采纳
请原谅我的无知，因为我对pyspark和Spark还不熟悉。我正在将Spark从1.6.3升级到2.1，在使用pyspark运行我们的模型时遇到了问题。
在运行抛出错误的python脚本时，我们所要做的就是读入一个JSON，并使用如下所示将其转换为DF
df_read = sparkSession.read.json('path to json file')
在这个读取之后，我们在DF上执行一些操作，在列上运行一些UDF，然后最终希望写回JSON，然后JSON将被提取并写入Apache Phoenix表。
当尝试在DF上执行任何终端操作时，例如show()或take()或任何类
浏览 8提问于2018-12-04得票数 0
在看过之后，我尝试使用下面的代码，但我知道答案中的createDataFrame与spark.read.json不同，不能以相同的方式使用。
import findspark
findspark.init('/home/myuser/softwareInstallations/spark-3.1.1-bin-hadoop2.7')
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import StructField, StructType, IntegerType, Stri
浏览 2提问于2021-04-01得票数 0
我正在使用Pyspark在Dataframe中转换JSON。我能够成功地改变它。但我面临的问题是，有一个键将出现在一些JSON文件中，而不会出现在另一个文件中。当我使用Pyspark SQL上下文展平JSON时，某些JSON文件中不存在键，它在创建Pyspark数据框时出错，抛出SQL分析异常。
例如，我的示例JSON
{
"_id" : ObjectId("5eba227a0bce34b401e7899a"),
"origin" : "inbound",
"converse" : "
浏览 0提问于2020-05-14得票数 1
我有一个如下所示的json文件： test= {'kpiData': [{'date': '2020-06-03 10:05',
'a': 'MINIMUMINTERVAL',
'b': 0.0,
'c': True},
{'date': '2020-06-03 10:10',
'a': 'MINIMUMINTERVAL',
'b': 0.0,
'c
浏览 24提问于2021-05-10得票数 0
回答已采纳
在PySpark和Synapse数据流中，我遇到了非常奇怪的错误。
我正在使用下面的查询读取JSON文件，但得到重复的列错误，即使没有重复的列。我可以使用其他工具和JSON验证器来读取它，也可以使用数据流，但不能在PySpark中读取。
PySpark查询如下：
df = (
spark.read.option("multiline", "true")
.options(encoding="UTF-8")
.load(
"abfss://<Container>]@<DIR>.
浏览 3提问于2021-11-25得票数 0
我需要将JSON的列表转换为pySpark DataFrames。JSON都有相同的架构。问题是JSON中dicts的值条目有不同的数据类型。
例如:字段complex是一个Dicts数组，Dict有四个键，但类型不同(整数、字符串、浮点数和嵌套Dict)。参见下面的示例JSON。
如果我使用df = spark.createDataFrame(json_list)从jsons中创建我的DataFrame，因为他无法正确地推断模式，所以pyspark“删除”了一些数据。PySpark决定complex-field的架构应为：StructType("complex", Array
浏览 23提问于2022-03-28得票数 0
回答已采纳
我正在尝试编写一些代码，使用Python 3.9.1中的pyspark (3.0.1)将JSON解嵌到Dataframe中。 我有一些虚拟数据，其模式如下： data.printSchema()
root
|-- recordID: string (nullable = true)
|-- customerDetails: struct (nullable = true)
| |-- name: string (nullable = true)
| |-- dob: string (nullable = true)
|-- familyMembers: array (n
浏览 38提问于2021-02-05得票数 2
回答已采纳
如何解析嵌套列表中的JSON字符串以激发pyspark中的数据帧？
输入数据框架：
+-------------+-----------------------------------------------+
|url |json |
+-------------+-----------------------------------------------+
|https://url.a|[[1572393600000, 1.000],[1572480000000, 1.007]]|
|h
浏览 1提问于2021-02-08得票数 2
回答已采纳
我正在尝试使用外部API将JIRA数据引入Foundry。当它通过Magritte传入时，数据被存储在AVRO中，并且有一个名为response的列。response列包含如下所示的数据... [{"id":"customfield_5","name":"test","custom":true,"orderable":true,"navigable":true,"searchable":true,"clauseNames":["cf[5]&#
浏览 9提问于2021-08-27得票数 2
回答已采纳
我正在尝试用这个(运行在Databricks上)将pyspark连接到MongoDB：
from pyspark import SparkConf, SparkContext
from pyspark.mllib.recommendation import ALS
from pyspark.sql import SQLContext
df = spark.read.format("com.mongodb.spark.sql.DefaultSource").load()
但我知道这个错误
java.lang.NoClassDefFoundError: org/apache/spa
浏览 2提问于2017-04-04得票数 1
回答已采纳
1回答
我尝试将以下data.json文件加载到星星之火数据文件中：
{"positionmessage":{"callsign": "PPH1", "name": 0.0, "mmsi": 100}}
{"positionmessage":{"callsign": "PPH2", "name": 0.0, "mmsi": 200}}
{"positionmessage":{"callsign": "PP
浏览 2提问于2020-05-18得票数 0
回答已采纳
1回答
如何用火花放电爆炸()来爆炸结构
、、、、
如何将以下JSON转换为后面的关系行？我所坚持的部分是，由于类型不匹配，pyspark explode()函数抛出了一个异常。我还没有找到一种将数据强制转换为适当格式的方法，这样我就可以在source对象的sample_json键中从每个对象中创建行。
JSON输入
sample_json = """
{
"dc_id": "dc-101",
"source": {
"sensor-igauge": {
"id": 10,
"ip": "
浏览 3提问于2021-02-10得票数 2
回答已采纳
1回答
我正在尝试使用pyspark读取一个avro文件，并根据特定的键对其中一个列进行排序。我的avro文件中的一个列包含一个MapType数据，我需要根据键进行排序。test只包含一行，实体列具有MapType数据。我的目的是将输出写回一个avro文件，但需要对键进行排序。不幸的是，我无法做到这一点，不确定这是否有可能在阿夫罗？它是以输入出现的相同方式写回的。下面是我的代码(我已经创建了一个笔记本来测试它)：
from pyspark.conf import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functi
浏览 4提问于2020-06-11得票数 2
1回答
我正在尝试用不同的模式读取由pyspark编写的avro文件。小数列精度的差异。下面是由pyspark编写的avro文件夹的文件夹结构
/mywork/avro_data/day1/part-*
/mywork/avro_data/day2/part-*
下面是它们的模式
day1 = spark.read.format('avro').load('/mywork/avro_data/day1')
day1.printSchema()
root
|-- price: decimal(5,2) (nullable = True)
day2 = spark.re
浏览 4提问于2020-07-07得票数 0
回答已采纳
我正在尝试从嵌套的JSON (具有动态模式)中提取某些参数，并使用pyspark生成火花数据。
我的代码非常适合于第1级(key:value)，但对于作为嵌套JSON一部分的每一对(key:value)都无法获得独立的列。
JSON模式示例
注意-这不是确切的模式。它只是给出了模式的嵌套性质的概念。
{
"tweet": {
"text": "RT @author original message"
"user": {
"screen_name": "Retweete
浏览 2提问于2019-04-28得票数 0
回答已采纳
当前pyspark格式的logFile，然后加载红移。
分析以json格式输出的关于logFile的每一项，添加一项，并将其加载到Redshift中。但是，对于每种类型，某些项的格式是不同的。(对于同一项，先应用Shcema。)即使输出，也会输入Escape字符。有没有一种方法可以动态地创建模式信息，并且输出jsonfile没有转义字符？
-环境--
- spark 2.4.0
- python version 2.7.15
- DataFrame --
>> df.printSchema()
root
|-- Name: string (nullable = false)
|
浏览 0提问于2018-12-28得票数 6
1回答
摘要
我需要在Camel服务器进行修改过程中构建一组统计信息，并将这些统计信息作为一个对象(一个json日志行)发出。这些统计数字需要包括：
输入文件度量(大小/字符/字节和其他文件部分的特定度量)
处理时间统计(开始/结束/处理时间持续时间，度量收集时间的开始/结束/持续时间)
输出文件指标(与输入文件度量相同，并将是不同的编号，输出文件正在更改)
输出文件度量是问题的，因为在文件写入磁盘之前我无法访问该文件，直到“进程”完成时才能将其写入磁盘
背景
log4j实现正在被用于服务日志记录，但经过一些修改之后，我们意识到它确实不适合这里的需求，因为它将输出多行jso
浏览 1提问于2019-06-20得票数 1
回答已采纳
我正在使用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
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
我有一个Json模式模板，其中包含对象数组。我需要使用该模板验证Json输入。但我希望这不应该依赖于数组中对象的顺序。
下面我们在模板中有3个不同对象的数组，即abs，端点和分派。我想从这里删除对顺序的依赖。我可以在Json输入模式中提供项目的顺序。它不应该依赖于模板。我使用'ajv‘node js模板来验证带有模板数据的Json输入。任何帮助都将不胜感激。谢谢。
Attached template and input json.
Json Template
var schema1 = {
"additionalProperties" : {
"
浏览 0提问于2018-10-09得票数 2
3回答
因此，我查看了一堆关于Pyspark、Jupyter和设置内存/核心/执行器(以及相关内存)的帖子。
但我好像被困住了-
问题1：我没有看到我的机器使用内核或内存。为什么？我是否可以对编译器/内核/内存做一些调整，以优化读取文件的速度？问题2:还有什么方法可以让我看到一个进度条，显示导入了多少文件ahs (火花监视器似乎做不到)。
我正在导入一个33.5gb的文件到火星雨。
机器有112 gb或RAM 8核/16个虚拟核。
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appNam
浏览 0提问于2020-08-13得票数 11
回答已采纳
5回答
我想读取pyspark.lf格式的json或xml文件我的文件被分成多行
rdd= sc.textFile(json or xml)
输入
{
" employees":
[
{
"firstName":"John",
"lastName":"Doe"
},
{
"firstName":"Anna"
]
}
输入分布在多行中。
预期输出{"employees:[{"firstName:"John",......]}
如何使用pyspa
浏览 2提问于2015-05-26得票数 10
我试着用spark连接到kafka的话题。它不会读取数据流中的任何数据，也不会产生任何错误。下面是我的jupyter代码：
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'
from pretty import pprint
from pyspark import SparkContext
from pyspark.streaming import Stream
浏览 0提问于2019-12-19得票数 1
亲爱的PySpark社区：
我想在供应前计算estimate_day_to_sustain。原始代码是使用‘however’语句用SAS编写的，但是，我无法在PySpark中找到解决它的方法。帮帮忙，谢谢！
输入数据：
产出数据：
算法：
第一天的estimate_day_to_sustain=current ：current
其他日子的：
如果1> estimate_day_to_sustain + estimate_day_to_sustain提供<= 当前日，则current estimate_day_to_sustain=current<cod
浏览 4提问于2022-07-18得票数 0
回答已采纳
我是PySpark的新手。
我有一个JSON文件，它在架构下面
df = spark.read.json(input_file)
df.printSchema()
|-- UrlsInfo: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- displayUrl: string (nullable = true)
| | |-- type: string (nullable = true)
| | |-- url: string (nul
浏览 0提问于2019-03-13得票数 0
回答已采纳
2回答
我是新来的火花，我正在尝试阅读json文件与火花。我得到的输出，如“当前输出”中提到的，正如我所需要的“预期输出”格式。
Json_file
{"Id":"665cfa8","Attributes":{"attr1":"0","attr2":"60","attr3":"","attr4":"p-queue"},"Channel":"CHAT","Method":"O
浏览 4提问于2022-04-22得票数 1
2回答
我从API中提取json数据。我的脚本将存储这些数据并将有关数据的信息添加到字典中。
为了存储json数据，我打算使用：
with open('data.json', 'w') as f:
json.dump(data, f)
怎样才是存放词典的合适方法？是否应该将dict转换为json格式？
json_str = json.dumps(dict1)
用和上面一样的方式保存它？
浏览 6提问于2016-02-18得票数 3
回答已采纳
7回答
我有一个由一个名为json的列组成的pyspark，其中每一行都是一个json的unicode字符串。我想解析每一行并返回一个新的dataframe，其中每一行都是解析的json。
# Sample Data Frame
jstr1 = u'{"header":{"id":12345,"foo":"bar"},"body":{"id":111000,"name":"foobar","sub_json":{"id":54321,&
浏览 25提问于2016-12-12得票数 56
回答已采纳
我转换了现有的代码，它在python中粘贴，下面是pyspark。
Python代码：
import json
import csv
def main():
# create a simple JSON array
with open('paytm_tweets_data_1495614657.json') as str:
tweetsList = []
# change the JSON string into a JSON object
jsonObject = json.load(str)
浏览 0提问于2017-06-18得票数 1
2回答
输入数据-
{"driverId":1,"driverRef":"hamilton","number":44,"code":"HAM","name":{"forename":"Lewis","surname":"Hamilton"},"dob":"1985-01-07","nationality":"British","url":"h
浏览 3提问于2021-10-02得票数 0
回答已采纳
我尝试通过schema_of_json函数从类似json的字符串中推断出一个模式，然后使用该模式通过from_json函数将该字符串值格式化为结构。我的代码是
import pyspark.sql.functions as sqlf
dfTemp = readFromEventHubs()
df= dfTemp.withColumn("col1", sqlf.get_json_object(col("jsonString"), '$.*'))
col1Val= df.col1
jsonSchema = sqlf.schema_of_json(co
浏览 0提问于2019-12-04得票数 2
我有一个dataframe，它有一个名为“特征”的列，它是一个由多个标志组成的整数。 我需要将此列转换为字符串列表(用于弹性搜索索引)。转换看起来像这样。 TRAIT_0 = 0
TRAIT_1 = 1
TRAIT_2 = 2
def flag_to_list(flag: int) -> List[str]:
trait_list = []
if flag & (1 << TRAIT_0):
trait_list.append("TRAIT_0")
elif flag & (1 << TRA
浏览 23提问于2020-04-28得票数 0
回答已采纳
2回答
我在Databricks中有一个表，其中有一个列作为字符串字典，如下所示- +---+--------------------------------------------------------------------------------------------------------------+
|id |stringDictionary |
+---+----------------------
浏览 23提问于2021-10-02得票数 3
回答已采纳
我每天都收到带有10个属性的json，但有些日子，如果任何属性没有值，他们就会发送9个属性，而第10个属性在json中没有。如何在不更改旧表模式的情况下读取pyspark中的json文件
浏览 8提问于2019-05-26得票数 0
1回答
火花流从阴阳石中回收信息的问题
、、、、
我有一个问题，我试着从GCP上的星星团实时接收来自pubsublite的消息，但是它们被分组在一个分钟的块中。
我的代码：
producer.py
import random
import time
from proj_BOLSA import settings
from google.cloud.pubsublite.cloudpubsub import PublisherClient
from google.cloud.pubsublite.types import (
CloudRegion,
CloudZone,
MessageMetadata,
Top
浏览 7提问于2022-08-05得票数 0
2回答
我有一个简单的JSON文件：
{"adas":{"parkAssist":{"rear":{"alarm":false,"muted":false},"front":{"alarm":false,"muted":false}},"lane":{"keepAssist":{"right":false,"left":false}}}}
但当我这样读的时候：
spark.read.option("inf
浏览 6提问于2022-06-08得票数 2
1回答
我想爆炸一个嵌套的json到CSV文件。希望将嵌套的json解析为行和列。
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext
from pyspark.sql.types import *
from pyspark.sql import functions as F
from pyspark.sql import Row
df=spark.read.option("multiline","true").json("sample1.json")
df.pr
浏览 7提问于2021-11-18得票数 0
回答已采纳
在我的应用程序中，我从S3上不同位置的数据创建不同的数据帧，然后尝试将这些数据帧合并为单个数据帧。现在，我正在使用一个for循环。但我有一种感觉，使用pyspark中的map和reduce函数可以更有效地完成这项工作。下面是我的代码：
from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext, GroupedData
import pandas as pd
from datetime import datetime
sparkConf = SparkConf().setAppName(
浏览 0提问于2016-08-24得票数 3
我在我的Linux计算机上使用PySpark。我的火花版本是2.4.4。
我有一个小脚本来初始化基本入口点，包括SparkContext、SQLContext和SparkSession。这是密码。
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
import pyspark.sql.functions as sqlfunc
--我不想每次打开PySpark时都输入这个。因此，我想要
( a)在我的终端中运行这个脚本
( b)继续在我的PySpark交互式shell中工作
，我该怎么做？
我阅读了下面的线程，以学习如
浏览 2提问于2019-10-18得票数 0
回答已采纳
我使用以下命令在Spark2.3集群中运行一个PySpark作业。
spark-submit
--deploy-mode cluster
--master yarn
--files ETLConfig.json
PySpark_ETL_Job_v0.2.py
ETLConfig.json有一个传递给PySpark脚本的参数。我在主块中引用这个config json文件，如下所示：
configFilePath = os.path.join(SparkFiles.getRootDirectory(), 'ETLConfig.json')
with open(configF
浏览 0提问于2018-09-16得票数 2
回答已采纳
1回答
看来，标准的Apache读取器/编写器只能基于NiFi模式解析JSON输入。
Avro模式是对JSON的限制，例如它不允许以数字开头的有效JSON属性。
JoltTransformJSON处理器在这里可以提供帮助(它没有对输入JSON的样子施加Avro限制)，但是这个处理器似乎不支持批处理FlowFiles。它也不是建立在读者和作家的基础上的(也许正因为如此)。
是否有方法读取任意有效的批处理JSON输入，例如以多行形式
{"myprop":"myval","12345":"12345",...}
{"myprop"
浏览 0提问于2018-11-14得票数 0
我正在使用pyspark从mutli行json对象中提取数据。我能够在文件中读取，但无法解析几何图形列的内容。
整个表的一个示例如下所示。
+--------------------+--------------------+-------+
| geometry| properties| type|
+--------------------+--------------------+-------+
|{[13.583336, 37.2...|{AGRIGENTO, AGRIG...|Feature|
|{[13.584538, 37.3...|
浏览 2提问于2021-11-11得票数 1
如何解析和转换json字符串从火花公子数据行？
我在寻找如何解析的帮助：
从json到output 1transform json字符串到列a、b和id output 2的json字符串
背景:我通过API字符串获得了大量行(jstr1、jstr2、.)，这些行被保存为激发df。我可以分别读取每一行的模式，但这不是解决方案，因为它非常慢，因为模式有大量行。每个jstr都有相同的模式，列/键a和b保持不变，只有id和列中的值变化。
编辑:黑主教解决方案使用MapType模式像魅力schema = "map<string, array<struct<a:int,b:int
浏览 1提问于2021-01-29得票数 2
回答已采纳
我试图提供一个文件列表，以便在需要的时候读取这些文件(这就是为什么我不希望使用boto或其他任何方法将所有文件预下载到实例中，然后才将它们读入spark“本地”)。
os.environ['PYSPARK_SUBMIT_ARGS'] = "--master local[3] pyspark-shell"
spark = SparkSession.builder.getOrCreate()
spark.sparkContext._jsc.hadoopConfiguration().set('fs.s3.access.key', credential
浏览 9提问于2022-02-27得票数 0
我正在尝试使用JSON模式验证json输入，但它并不像我所需要的那样工作。
我有以下输入JSON (其中一部分)：
[
{
"admin_state": "disabled"
},
{
"state": "disabled"
}
]
以及以下json模式(也是其中的一部分)：
{
"type": "array",
"items": [
{
"type": "object",
浏览 0提问于2018-03-16得票数 7
回答已采纳
云服务器
ICP备案
实时音视频
对象存储
云直播
腾讯云开发者
