我在Azure技术上工作，想要读取json文件，它超过3 3GB，并且在嵌套元素中有重复的列。我试过pyspark，数据流，pipeines。但不走运。你能建议我使用哪种技术吗？
浏览 34提问于2021-11-18得票数 0
我有一个100k行，20M列的pyspark.mllib.linalg.SparseMatrix。在PySpark中将其保存为稀疏矩阵的最佳格式是什么？
我发现另一个帖子的主题与一年前略有不同，但答案并不能解决我的问题。。
浏览 31提问于2018-06-05得票数 1
回答已采纳
我有一个巨大的json文件35-40GB大小，它是hdfs上的一个多行JSON。我已经使用了.option('multiline', 'true').read.json('MULTILINE_JSONFILE_.json').repartition(50)和Pyspark。
我已经弹出了60个执行器、16个内核、16 up的Ememory和设置了内存开销参数。每跑一次，执行者都会迷路。
它可以很好地工作在较小的文件中，但不适用于文件> 15 GB，我有足够的集群资源。
从星星之火UI中，我看到的是，每当数据被单个执行器处理时，所有其他执行器
浏览 1提问于2019-08-17得票数 2
我有一张这样的桌子：
row | key | json
0 | a | {'something':{'a':1}}
1 | b | {'something':{'a':2, 'b':3 }}
我想检索在key列中定义的json中的值：
row | value
0 | 1 // from $.something.a
1 | 3 // from $.something.b
在蜂巢上你可以做
select get_json_object(json, concat("
浏览 1提问于2018-06-28得票数 0
2回答
电火花迭代列加内存泄漏
、、、
我一直在尝试对pyspark数据执行一些迭代计算。列是根据前面的列添加到df中的。然而，我注意到，使用的内存不断增加。下面是一个简单的例子。
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
from pyspark import Row
conf = SparkConf().setAppName("myFirstApp").setMaster("local")
sc = SparkContext(conf=conf)
sqlContext = SQLC
浏览 3提问于2019-10-21得票数 0
回答已采纳
我们一直在探索如何使用Glue将一些JSON数据转换为parquet。我们尝试过的一种情况是在拼花表中添加一列。所以分区1有A列，分区2有A，B列，然后我们想编写进一步的Glue ETL作业来聚合parquet表，但是新列不可用。使用glue_context.create_dynamic_frame.from_catalog加载动态框架，我们的新列从未出现在模式中。
我们为我们的爬行器尝试了几种配置。对所有分区使用单一架构，对s3路径使用单个架构，对每个分区使用架构。我们总是可以在Glue表数据中看到新列，但是如果我们使用pyspark从Glue作业中查询它，那么它总是空的。当我们下载一些样本
浏览 4提问于2019-04-09得票数 8
我在GCP中启动了一个DataProc集群，有一个主节点和3个工作节点。每个节点有8个vCPU和30G内存。
我开发了一个pyspark代码，它从GCS读取一个csv文件。csv文件的大小约为30G。
df_raw = (
spark
.read
.schema(schema)
.option('header', 'true')
.option('quote', '"')
.option('multiline',
浏览 4提问于2021-05-03得票数 0
我正在研究PySpark。我有一个数据框，我需要将其转储为JSON文件，但JSON文件应具有以下格式，例如- {"Column 1": [9202, 9202, 9202, ....], "Column 2": ["FEMALE", "No matching concept", "MALE", ....]} 因此，每列应该有一个键，相应的值应该包含该列中的所有值的列表 我尝试将其转换为Pandas数据帧，然后在将其作为JSON转储之前转换为字典，并成功地完成了转换，但由于数据量非常大，我想直接在PySpark数据
浏览 14提问于2019-12-18得票数 1
回答已采纳
1回答
我想知道这是否是Server (在Azure上)的正常行为。我的SQL查询使用返回JSON路径、和嵌套查询。我的问题是，似乎是，根据结果JSON的长度，结果被分隔成多个行。
这是一个问题，因为行数可能会根据数据变化。我还必须手动将所有行连接在一起才能获得有效的JSON。
解决方法是将生成的JSON放入变量中，并在手动之后选择它。这会导致一行和列(因此我知道这不是最大长度问题)。
这是正常行为还是我的查询有问题？
浏览 1提问于2017-01-26得票数 3
回答已采纳
我正在使用下面的pyspark代码从s3桶中读取数千个JSON文件
sc = SparkContext()
sqlContext = SQLContext(sc)
sqlContext.read.json("s3://bucknet_name/*/*/*.json")
这需要大量时间来读取和解析JSON文件(~16分钟)。我如何并行化或加速这个过程？
浏览 1提问于2018-06-20得票数 2
这篇文章很好地展示了如何使用pyspark ()将固定宽度的文本文件解析成Spark数据帧。
我有几个要解析的文本文件，但每个文件的模式都略有不同。而不是像上一篇文章所建议的那样，为每个文件编写相同的过程，我想编写一个通用函数，它可以解析给定宽度和列名的固定宽度文本文件。
我是pyspark的新手，所以我不确定如何编写列数和类型可变的select语句。
任何帮助都将不胜感激！
浏览 0提问于2017-09-08得票数 2
我正在使用AnyLogic来模拟一个进程。我发现很难对以下内容进行建模。想象一下装着产品的盒子的到来。这些产品(比如钢笔)有不同的尺寸。我在CSV文件中有数百个这样的盒子。我想把它们作为盒子带到系统中，然后我想根据它们的大小在模拟中拆分它们。 我是否需要一个接一个地创建这些笔，然后使用'batch‘模块将它们放入盒子中？或者我可以很容易地阅读带有盒子的csv，其中每个盒子都有很多笔？
浏览 13提问于2019-12-17得票数 0
回答已采纳
我有包含一些数据的json文件，我将这个json转换为pyspark dataframe(我选择了一些列，而不是所有列)，这是我的代码： import os
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession
import json
from pyspark.sql.functions import col
sc = SparkContext.getOrCreate()
spark = SparkSession
浏览 125提问于2019-03-12得票数 2
回答已采纳
1回答
我有一系列的技能。我正在尝试过滤POS_NNP列。如果word在skills中，则(word,1) else (word,0)，格式为(word1,1)，(word2,0)，(word1,1)....对于每一行。 技能中的一些单词是机器学习，人工智能，Python(编程语言)…… 我已经创建了一个if-else，但它抛出了错误，我不知道如何修复它。 sample['skill/notskill']=sample['POS_NNP'].apply(lambda x:','.join([(x,1) if x in skill else (x, 0)
浏览 14提问于2021-08-17得票数 0
回答已采纳
我正在尝试使用外部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
回答已采纳
我正在将json文件导入到pyspark dataframe中。我已经用以下代码导入了json df = sqlContext.read.json("json_file.json").select("item", "attributes") 我想将属性从一列拆分为多列。 以下是json格式示例： {"item":"item-1","attributes":{"att-a":"att-a-15","att-b":"att-b-10",
浏览 12提问于2020-01-27得票数 0
1回答
-编辑-
这个简单的例子只显示了3条记录，但我需要对数十亿条记录这样做，所以我需要使用Pandas，而不是仅仅将Spark转换为Pandas并使用简单的应用程序。
输入数据
期望输出
-END编辑-
我一直把头撞在墙上试图解决这个问题，我希望有人能帮我解决这个问题。我正在尝试将PySpark数据中的纬度/经度值转换为优步的H3十六进制系统。这是对函数h3.geo_to_h3(lat=lat, lng=lon, resolution=7)的非常直接的使用。但是，我的PySpark集群一直存在问题。
我正在使用以下命令设置PySpark集群，如databricks文章中所述：
浏览 6提问于2021-07-15得票数 2
回答已采纳
我的数据看起来就像
|ID|Notes|
---------------
|1|'{"Country":"USA","Count":"1000"}'|
|2|{"Country":"USA","Count":"1000"}|
ID : int
Notes : string
当我使用from_json解析列Notes时，它会给出所有的Null值。我需要帮助将本列Notes解析为pyspark中的列
浏览 7提问于2022-05-20得票数 2
在PySpark中，当我想要检查其中一个值是否在列中时，我使用.isin() F.col('metric').isin({"Metric_value1, Metric_value2"}) 像下面这样对单个值执行相同的检查是否正确，或者有更好的方法这样做？ F.col('metric').isin({"Metric_value1"})
浏览 21提问于2020-06-18得票数 1
回答已采纳
2回答
我有接口请求链接-
我需要更新google Sheets中的数据，在Google Sheets的一个列中使用ID，在另一列中使用国家名称
前两个ID的JSON输出如下所示
[
{
"ID": "AF",
"LocalizedName": "Afghanistan",
"EnglishName": "Afghanistan"
},
{
"ID": "AM",
浏览 13提问于2021-05-19得票数 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
我是PySpark的新手。
我只想保留至少有10个值的列
我已经使用describe来获取每列的非空记录的计数
现在如何提取值小于10的列名，然后在写入新文件之前删除这些列
df = spark.read.parquet(file)
col_count = df.describe().filter($"summary" == "count")
浏览 1提问于2019-09-28得票数 1
1回答
我使用aws glue来转换dynamodb表中的数据。我能够获取数据并使用pyspark逻辑进行转换。但是当我尝试将文件写入json格式时，我得到了一个无效的json格式.。在转换这个json文件时需要帮助。
代码：
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from pyspark import
浏览 10提问于2020-05-19得票数 0
1回答
有人能帮助解决这个问题吗？它没有显示名为'numpy‘的模块，但我已经在笔记本中安装并导入了它。 Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 223.0 failed 1 times, most recent failure: Lost task 0.0 in stage
浏览 118提问于2019-12-03得票数 0
2回答
我一直在搜索，还没有找到一种方法来重构dataframe的列，以便根据数组内容动态地向dataframe添加新的列。我对python还不熟悉，所以我可能在错误的条件下搜索，这也是我还没有找到一个明确的例子的原因。请让我知道，如果这是一个重复和参考链接，以找到它。我想我只是需要被指引到正确的方向。
好的，细节。
这里的环境是pyspark 2.3.2和python 2.7。
示例列包含两个数组，它们之间是相互关联的1到1。我想为title数组中的每个值创建一个列，并将相应的名称(在person数组中)放在相应的列中。
我拼凑了一个例子，集中在我的问题上，改变了数据。
import json
fr
浏览 1提问于2018-11-03得票数 0
回答已采纳
如何使用PySpark读取下面的JSON结构来触发数据帧？
我的JSON结构
{"results":[{"a":1,"b":2,"c":"name"},{"a":2,"b":5,"c":"foo"}]}
我已经尝试过了：
df = spark.read.json('simple.json');
我希望输出a，b，c作为列，值作为各自的行。
谢谢。
浏览 5提问于2018-03-21得票数 4
回答已采纳
我正在使用pyspark从mutli行json对象中提取数据。我能够在文件中读取，但无法解析几何图形列的内容。
整个表的一个示例如下所示。
+--------------------+--------------------+-------+
| geometry| properties| type|
+--------------------+--------------------+-------+
|{[13.583336, 37.2...|{AGRIGENTO, AGRIG...|Feature|
|{[13.584538, 37.3...|
浏览 2提问于2021-11-11得票数 1
我面临着一个奇怪的问题，我试图显示我的JSON对象的值，它在select()中运行得很好，但是它不适用于selectExp()，我得到了一个奇怪的错误，在我的实现中，
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("JsonPractice").getOrCreate()
my_json_df = spark.range(1).s
浏览 0提问于2020-07-09得票数 0
我有一个这样的csv文件： "request"
"{""CustomerId"":""1"",""EffectiveTime"":""2021-07-30T12:00""}"
"{""CustomerId"":""2"",""EffectiveTime"":""2021-07-30T13:00""}&
浏览 45提问于2021-09-22得票数 0
1回答
按列加权平均
、、
例如，我有这样一个数据集
test = spark.createDataFrame([
(0, 1, 5, "2018-06-03", "Region A"),
(1, 1, 2, "2018-06-04", "Region B"),
(2, 2, 1, "2018-06-03", "Region B"),
(3, 3, 1, "2018-06-01", "Region A"),
(3, 1, 3, "2018-06-0
浏览 2提问于2018-09-09得票数 3
1回答
我的工作表上有多列。为了前夫。我的列名是- a1，a2，a3，b1，b2，b3，c1，c2，c3.我想要创建100%的堆叠列图，其中a1，a2，a3应该在图表的同一1条，然后b1，b2，b3在图表的第二栏，c1，c2，c3应该在图表的第3栏。
我所做的是，我选择了100%堆叠柱状图，并选择了a1，a2，a3在y轴。直到现在为止，它的工作，然后，当我想把b1，b2，b3放在另一个酒吧，它不工作，这些值显示在同一条。我试着分组，但不能分组多列。
浏览 5提问于2022-08-16得票数 0
回答已采纳
1回答
在PySpark中给出的ALS示例中，根据本文档-- )，使用的数据在一列中具有显式反馈。这些数据是这样的:收人品的收视率高
然而，在我的例子中，我在多个列中有隐式反馈，如:用以下几个列进行隐式反馈：\{e76f}
我知道我们可以通过将implicitPrefs设置为False来使用隐式反馈。但是，它只接受一列。如何使用多列？
我发现了一个问题：，但是它与星火法和交替最小二乘法无关。我是否必须根据这个答案手动分配加权方案？或者在PySpark中是否有更好的解决方案？
浏览 5提问于2022-06-17得票数 0
1回答
我使用的是PySpark，我需要处理附加到单个数据帧中的日志文件。大多数列看起来都是正常的，但其中一列在{}中有JSON字符串。基本上，每一行都是一个单独的事件，对于JSON字符串，我可以应用单独的Schema。但我不知道在这里处理数据的最好方法是什么。 示例： ? 这个表稍后将帮助我以所需的方式聚合事件。 我尝试使用函数withColumn和from_json。它成功地在单个列中工作： from pyspark.sql.types import *
import pyspark.sql.functions as F
df = (df
.withColumn("n
浏览 15提问于2021-09-01得票数 0
我正在尝试使用SPARK-SQL和pyspark包装器来实现。我遇到了这个错误
pyspark.sql.utils.AnalysisException: u"cannot resolve '```(qtr)?+.+```'
given input columns:
这是我的query..Basically试图排除列'qtr‘。
select `(qtr)?+.+` from project.table;
使用以下属性在hive/beeline中工作得很好
set hive.support.quoted.identifiers=none;
有什
浏览 1提问于2018-09-27得票数 3
下面的脚本(Spark1.6)使用java.lang.NullPointerException中止，主要是由于函数滞后。请给我建议。
from pyspark.sql import HiveContext
sqlc= HiveContext(sc)
rdd = sc.parallelize([(1, 65), (2, 66), (3, 65), (4, 68), (5, 71)])
df = sqlc.createDataFrame(rdd, ["account_nbr", "date_time"])
df.registerTempTable("t
浏览 7提问于2017-06-10得票数 0
我对pyspark和json解析还是个新手，我被困在了某些特定的场景中。让我先解释一下我要做什么，我有一个json文件，其中有一个数据元素，这个数据元素是一个包含另外两个json对象的数组。给定的json文件如下所示
{
"id": "da20d14c.92ba6",
"type": "Data Transformation Node",
"name": "",
"topic": "",
"x": 380,
浏览 142提问于2019-06-05得票数 0
回答已采纳
我正在寻找使用实体框架核心将字符串列表存储到数据库中的最佳实践。
我的数据是一组角色，可能大约100个或更多，所以我有两个选项：
使用另一个表并建立一些关系。
通过nvarchar(max)使用一列类型的;和单独的角色
我需要知道哪条路更好，为什么？
现在，我的代码使用选项2来存储数据和实体框架值转换器，以便使用;分隔符将列表转换为字符串，反之亦然。
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
base.OnModelCreating(modelBuilder);
m
浏览 1提问于2018-08-27得票数 0
回答已采纳
2回答
我有一个核心数据管理对象，它有一个"Boolean“类型的属性。
在我的头文件中，我得到了：@property (非原子，保留) NSNumber * includeInHistory；
我使用的是@dynamic includeInHistory实现
在保存到磁盘之前与该托管对象的实例交互时，我得到了通过NSNumber接口实际映射到NSCFBoolean的内容。我使用"json-framework“来编码一些字典，其中包含来自Core Data的值。
问题是，在保存和检索数据之后，includeInHistory返回的似乎是一个标准的NSNumber (整数，不是Boole
浏览 5提问于2009-09-29得票数 1
我正在使用pyspark中的to_json将dataframe中的结构列转换为json列，但是在json中忽略了少数结构字段中的空值，我不希望这些空值被忽略。
浏览 6提问于2020-10-14得票数 0
我在S3 location s3://sample/input_data中有我所有的数据
我通过部署AWS并使用PySpark来完成ETL。
PySpark脚本非常简单。
我加载s3://sample/input_data作为火花数据。
将其划分为一列。
将dataframe保存为Parquet文件，在“追加”模式下使用写选项将其保存到S3 location s3://sample/output_data中
然后将s3://sample/input_data中的所有文件复制到s3://sample/archive_data并删除s3://sample/input_data中的所
浏览 3提问于2021-11-17得票数 0
回答已采纳
首先，我要预先感谢大家的帮助！我有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
我正在构建一个数据管道，它使用json格式的RESTApi数据并推送到Spark Dataframe。Spark版本: 2.4.4 但是得到的错误是 df = SQLContext.jsonRDD(rdd)
AttributeError: type object 'SQLContext' has no attribute 'jsonRDD' 代码： from pyspark import SparkConf,SparkContext
from pyspark.sql import SparkSession
from urllib import urlopen
浏览 27提问于2020-06-24得票数 2
回答已采纳
我有一个名为ARRIVAL DELAY的列，根据它的值，我必须编写一个python函数来自动执行数据标记。例如。小于5为早(0)，5到20为准时(1)，大于20为晚(2)。但该函数必须提前、按时和延迟地确定每个类别的范围(例如，使用bin大小)我的代码目前是硬编码的。 from pyspark.sql.functions import when,col,lit
flightsDf = flightsDf.withColumn("multiClassArrDelay", when((flightsDf.ARRIVAL_DELAY < 5), lit(0)).when((
浏览 34提问于2021-09-12得票数 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
1回答
我正在尝试读取从PySpark.My到JSON的流JSON数据，如下所示：
{'installmentNo': '10', 'loanId': '1'}
我已经指定了模式，但是当spark读取数据时，我会得到'null‘。下面是代码片段。
from pyspark.sql.types import *
from pyspark.sql.functions import from_json
fields = [
StructField("installmentNo", IntegerType(), T
浏览 0提问于2018-09-21得票数 0
回答已采纳
我有一个数据帧 import os, sys
import json, time, random, string, requests
import pyodbc
from pyspark import SparkConf, SparkContext, SQLContext
from pyspark.sql.functions import explode, col, from_json, lit
from pyspark.sql import functions as f
from pyspark.sql import SparkSession
from pyspark.sql.types
浏览 28提问于2020-10-06得票数 0
回答已采纳
我对pyspark完全是个新手。我在执行命令时遇到错误
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("basics").getOrCreate()
df = spark.read.csv("data.csv",inferSchema=True,header=True)
df.columns
我的数据有1,000,000行和50列。我得到了下面的错误。
ValueError Traceback (
浏览 0提问于2019-10-02得票数 0
1回答
我有一个Hbase表：“员工详细信息”和列族：“个人详细信息”，“专业详细信息”
当我试图为多个列族编写使用put时，如下所示：
put 'employeedetails','1','personaldetails:firstname','Steven','personaldetails:lastname','Gerrard','professionaldetails:company','ABC','professionaldetails:empid'
浏览 5提问于2017-03-31得票数 1
回答已采纳
1回答
我有一个python类，它的函数如下：
class Features():
def __init__(self, json):
self.json = json
def get_email(self):
email = self.json.get('fields', {}).get('email', None)
return email
我正尝试在一个pyspark数据帧中使用get_email函数，基于另一个列"raw_json“创建一个新列，该列由json值组成：
df = data.
浏览 6提问于2020-11-12得票数 0
回答已采纳
1回答
我有一个嵌套的json，它有一个键，值是另一个带有key和value的json。下面是json。
{
"Testicular Torsion": {
"What is testicular torsion?": "ABC",
"Symptoms": "AB",
"Risks": "AL",
"Diagnosis
浏览 1提问于2018-06-21得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
