PySpark。在读取拼图时通过强制转换为字符串合并架构？

文章/答案/技术大牛

发布

1回答

python、apache-spark、pyspark

我从拼图文件中读取数据，其中有一个Map type字段，如下所示：>>> df.collect() |-- key: string问题是在其他分区中，密钥a为None，导致密钥b被读取为) field: map (contains

浏览 11提问于2019-10-23得票数 1

1回答

关于在PySpark中写入拼图的问题

csv、pyspark、parquet

在PySpark中将csv文件转换为parquet时遇到问题。当转换相同模式的多个文件时，它们不具有相同的模式，因为有时数字字符串将被读取为浮点型，其他字符串将被读取为整数，等等。似乎当编写具有相同列的数据帧时，但是以不同的顺序排列以拼接，那么这些拼接就不能被加载到相同的语句中。如何将dataframe写入parquet，以使所有列都以字符串类型存储？如何处理列的顺序？在写入<e

浏览 23提问于2021-01-20得票数 0

2回答

如何使用Spark df.write.parquet编写具有已定义模式的拼图文件。-火花源

json、apache-spark、dataframe、pyspark、parquet

我正在使用df.write将JSON转换为拼图文件。在我的JSON文件中，我所有的列都是字符串，所以在读取数据帧时，我使用schema来推断，文件中列的数目也在不断变化。当写到parquet时，我不想把它们写成字符串，而是想把一些列改成date和decimal。我知道在捕获到另一个df之前，我们可以在df select中选择并在列级中进行强制转换。示例： JSON读取</e

浏览 0提问于2018-08-10得票数 0

3回答

Spark读取不同版本的拼图文件

apache-spark、parquet、versions

我用Version1模式生成了一年多的拼图文件。对于最近的模式更改，较新的parquet文件具有Version2模式额外的列。因此，当我从旧版本和新版本一起加载拼图文件，并尝试过滤更改的列时，我得到了一个异常。我想让spark读取旧文件和新文件，并在列不是present.Is的地方填充空值。有没有解决方法，当找不到列时，spark填充空值？

浏览 3提问于2017-04-28得票数 2

1回答

如何使用pyspark和自定义python函数处理均匀的to流

apache-spark、pyspark、azure-eventhub

我当前的设置是：我使用作为如何读取数据的示例，但是：获取流的每个元素并通过python函数传递它的正确方法是什

浏览 52提问于2018-03-19得票数 3

1回答

如何解析星火中不同类型的JSON列表？

json、dataframe、apache-spark、pyspark、apache-spark-sql

changes": [["buy", "0.924000", "34100.21"], ["sell", "0.928000", "101.1"]]} 我需要为它定义一个模式，并在FAILFAST模式下读取它

浏览 6提问于2022-02-11得票数 0

3回答

如何在pyspark数据框中将字符串类型的列转换为int形式？

python、dataframe、pyspark

我在pyspark有数据帧。它的一些数字列包含'nan‘，所以当我读取数据并检查dataframe的模式时，这些列将具有'string’类型。如何将它们更改为int类型。我将'nan‘值替换为0，并再次检查了架构，但同时也显示了这些列的字符串类型。data_df.fillna(0)我的数据如下所示：在这里，包含整数值的“Plays”和“drafts”列，但由于这些列中存在nan，因此它们

浏览 2提问于2017-10-26得票数 64

2回答

使用pyspark重新分区失败并出现错误

pyspark、apache-spark-sql、parquet

我有实木地板在s3文件夹与下面的镶木地板的column.Size大约是40MB。，我正在使用pyspark从s3读取文件并将其写入s3存储桶。spark_df.repartition(1).write.partitionBy(partition_columns).mode('append').parquet(file_path_re)org_id, device_id, channel_id

浏览 0提问于2020-04-29得票数 1

2回答

在pySpark中处理空数组(可选二进制元素(UTF8)不是一个组)

python、apache-spark、pyspark

df = spark.read.parquet(bad_partition_path)some-array: array当然，如果我想读取多个分区，这是一个问题，因为schemas不能合并模式。element: struct array-field-2: string 到目前为止还不错，但是当我试图实际收集数据时，架构不应导致不兼容。如果您想知道，在不指定模式的情况

浏览 8提问于2020-02-19得票数 5

回答已采纳

3回答

使用spark读取不存在的带有Parquet格式的列

python、apache-spark、pyspark、apache-spark-sql、parquet

6125|SQL显示错误：我知道原因是2017_01.parquet没有vnum列，我有两个know解决方案，一个是使用mergeSchema，另一个是在读取parquet文件时

浏览 1提问于2017-09-25得票数 4

1回答

似乎无论文件是如何编写的，parquet总是将文件的模式转换为可空的列。在读取这些文件时，我希望拒绝在特定列中包含空值的文件。如果您尝试在拼花，它将被接受。我可以对空值的列进行筛选或计数，然后引发错误-从性能角度来看，这是很糟糕的，因为我将在工作中获得额外的阶段。有没有办法在读取时强制对文件进行验证？如果有帮助的话，我使用的是Spark 3版本。使用示例进行编辑： from pyspark.sql.types import *

浏览 29提问于2020-09-25得票数 0

回答已采纳

1回答

如何向Spark中的现有分区添加行？

apache-spark、amazon-s3、pyspark

当前分区是通过date：created_year={}/created_month={}/created_day={}实现的。我还读到数据帧在Spark中是不可变的。那么，如何实现在新数据到达现有分区时将其附加到现有分区并每天维护一个对象呢？

浏览 0提问于2020-08-05得票数 2

1回答

为什么OpenCL工作组规模对GPU性能有很大影响？

c++、caching、opencl、gpgpu、memory-access

我正在Qualcomm Adreno 630 GPU上对一个简单的矩阵转置内核进行基准测试，我试图了解不同工作组规模的影响，但令人惊讶的是，我得到了一些有趣的结果，我无法解释。我知道内存合并或缓存可以在这里发挥作用，但我不能完全解释这一点。

浏览 35提问于2020-08-07得票数 1

1回答

读取parquet文件时，无法解析包含int和double的合并方案

scala、apache-spark、apache-spark-sql

我有两个拼图文件，一个包含一个整型字段myField，另一个包含一个双精度字段myField。当尝试一次读取两个文件时val fileWithInt = basePath + "intFile.snappy.parquet"在传递显式<em

浏览 0提问于2018-12-18得票数 10

1回答

如何从Pyspark中读取列并在其上应用UDF？

apache-spark、pyspark

我正在创建一个DF，方法是读取Pyspark中的csv文件，然后转换为RDD来应用UDF。它在应用UDF时抛出一个错误。(col2,StringType,true),StructField(col3,StringType,true)))文件跟踪(最近一次调用)：文件""，第1行，在"/usr/lib/spark/python/pysp

浏览 2提问于2020-02-21得票数 0

回答已采纳

3回答

蜂巢兽人返回零

hive、null、emr、orc

我在AWS上使用EMR实例。我是否应该配置/检查以支持蜂巢的ORC格式？

浏览 7提问于2017-10-24得票数 1

1回答

火花DataFrame读写

scala、apache-spark、hive

由于我正在将数据从半结构化格式转换为结构化格式，但我的模式略有变化，因此我认为像json这样的数据源的inferSchema选项将对我有所帮助。spark.read.option("inferSchema","true").json(RDD[String])案例1:对于较小的数据，所有的拼花文件都有双倍的数量我试着调试，并发现了一些概念，比如模式进化和模式合并，这让我产生了更多的疑问，而不是答案。我的怀疑/问题

浏览 0提问于2018-11-18得票数 1

回答已采纳

2回答

使用dask合并csv文件

python、dask

我正在使用dask读取5个大的(>1 GB) csv文件，并将它们合并(类似SQL)到一个dask数据帧中。现在，我正在尝试将合并后的结果写入单个csv。我在dask dataframe上使用compute()将数据收集到单个df中，然后调用to_csv。但是，compute()在读取所有分区的数据时速度很慢。我尝试在dask df上直接调用to_csv，它创建了多个.part文件(我没有尝试将这些.part文件合并到一个csv中)。

浏览 5提问于2017-03-23得票数 4

1回答

火花动态分区:读取时的SchemaColumnConvertNotSupportedException

apache-spark

问题我使用PySpark v2.4.5、拼花格式和动态分区，并采用以下层次结构：BASE_PATH/COUNTRY=US/TYPE=sms/YEAR=2020/MONTH=04/DAY=10/。这是因为不同类型(即sms和mms之间)的架构不同。看起来像星火试图合并到模式读取下的遮罩下。更准确地说，我可以读取F.col('TYPE') == &

浏览 0提问于2020-04-15得票数 0

1回答

如何解析混合嵌套和非嵌套结构的json？

json、scala、apache-spark、nested、apache-spark-sql

在文件1中，JSON元素"image“是嵌套的。true) |-- type: string (nullable = true){"id": "0002", "type": "donut", "name": "CupCake", "i

浏览 0提问于2017-10-29得票数 1

回答已采纳

点击加载更多