使用pyspark对csv文件中的100行进行分区
，可以按照以下步骤进行：
- 导入必要的库和模块：
from pyspark.sql import SparkSession
from pyspark.sql.functions import monotonically_increasing_id- 创建SparkSession对象：
spark = SparkSession.builder.appName("CSV Partitioning").getOrCreate()- 读取CSV文件并创建DataFrame：
df = spark.read.csv("path/to/csv/file.csv", header=True, inferSchema=True)其中，"path/to/csv/file.csv"是CSV文件的路径，header=True表示第一行是列名，inferSchema=True表示自动推断列的数据类型。
- 添加行索引列：
df = df.withColumn("row_id", monotonically_increasing_id())这一步是为了给每一行添加一个唯一的行索引列，方便后续分区操作。
- 对DataFrame进行分区：
partitioned_df = df.repartitionByRange(10, "row_id")这里使用repartitionByRange方法将DataFrame分成10个分区，"row_id"是用于分区的列名。
- 获取分区后的DataFrame的前100行：
result_df = partitioned_df.limit(100)使用limit方法获取前100行数据。
- 展示结果：
result_df.show()这一步是将结果展示出来，可以根据需要进行进一步处理或保存。
以上是使用pyspark对csv文件中的100行进行分区的步骤。在这个过程中，pyspark提供了强大的分布式计算能力，可以处理大规模的数据集。分区可以提高数据处理的效率和并行性。对于更复杂的操作，还可以使用pyspark的其他功能和API进行数据处理和分析。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark：https://cloud.tencent.com/product/spark
- 腾讯云数据仓库（TencentDB for TDSQL）：https://cloud.tencent.com/product/tdsql
- 腾讯云对象存储（COS）：https://cloud.tencent.com/product/cos
- 腾讯云大数据计算服务（Tencent Cloud Big Data）：https://cloud.tencent.com/product/tcbds
相关·内容
2回答
我正在学习吡咯烷酮，我对如何将分组数据文件保存为csv文件感到有点困惑(假设是出于某种原因-例如。RAM限制--我不想先把它转换成Pandas数据格式)。
作为一个可复制的例子：
import seaborn as sns
import findspark
findspark.init()
import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.master('local') \
.appName('Data cleaning') \
.getOrCr
浏览 2提问于2019-02-07得票数 0
回答已采纳
我在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
我已经编写了在本地运行的Spark代码。我已经创建了一个用户定义的函数，该函数需要应用于通过交叉连接从本地文件读取的两个表创建的数据帧。
不知何故，我正在应用的用户定义函数没有使用，它没有被分发。我已经在所有节点上安装了所需的python包。使用spark-submit，我已经指定了内核和内存的数量。这是我的代码：
spark = SparkSession.builder.appName("WordSimilarities").enableHiveSupport().getOrCreate()
spark.sparkContext.parallelize(range(1,100
浏览 2提问于2018-01-09得票数 1
我想知道下面的伪代码是否是从PySpark()读取存储在Azure数据湖中的日期范围之间的多个拼花文件的有效方法。注意:拼花文件没有按日期进行分区。
我使用uat/EntityName/2019/01/01/EntityName_2019_01_01_HHMMSS.parquet惯例在ADL中存储数据，正如Nathan在书“大数据”( Big )中所建议的，稍微做了修改(使用2019年而不是year=2019)。
使用*通配符读取所有数据：
df = spark.read.parquet(uat/EntityName/*/*/*/*)
添加列FileTimestamp，该列使用字符串操作从En
浏览 1提问于2019-02-28得票数 1
回答已采纳
1回答
我对AWS胶水和火花比较陌生。我想按S3中的用户ID对csv文件进行排序。我正在试用下面的脚本，但是它没有对file.Can进行排序--请有人帮我做这个？
import sys
import math
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 awsglue
浏览 3提问于2022-01-16得票数 0
1回答
在下面的代码中，out.csv是parquet格式的。对于它作为csv文件写入它，我没有什么选择？
import py4j
from pyspark import SparkConf, SparkContext
from pyspark import HiveContext as hc
import os
from pyspark.sql import SQLContext, Row
from datetime import datetime
from pyspark.sql.types import DateType,StringType
import pyspark.sql.functi
浏览 0提问于2017-06-29得票数 1
回答已采纳
6回答
我使用的是Spark 1.3.1 (PySpark)，并且我已经使用SQL查询生成了一个表。我现在有一个对象，它是一个DataFrame。我想把这个表对象(我把它叫做“DataFrame”)导出到一个csv文件中，这样我就可以操作它并绘制列。如何将DataFrame“表”导出为csv文件？
谢谢!
浏览 813提问于2015-07-13得票数 92
回答已采纳
我希望与PySpark并行处理几个类似大小(100 MB)的依赖于i的csv文件。我在一台机器上运行PySpark : spark.driver.memory 20g spark.executor.memory 2g local1
文件内容: type (在每个csv中具有相同的值)、时间戳、价格
首先，我在一个csv上测试了它(注意，我使用了35个不同的窗口函数)：
logData = spark.read.csv("TypeA.csv", header=False,schema=schema)
// Compute moving avg. I used 35
浏览 3提问于2020-08-23得票数 0
我在Databricks的Pyspark环境中工作，有一个pyspark数据框架，我将其称为df。 我需要将这个spark数据帧推送到csv文件中，我无法这样做。虽然没有弹出错误，但数据帧没有复制到csv中。下面是通用代码 path = “ “ #CSV File Location
header = “This is the header of the file"
With open(path,”a”) as f:
f.write(header+”\n”)
df.write.csv(path=path,format=“csv”,mode=“append”)
浏览 17提问于2020-06-21得票数 0
我正在使用PySpark中的以下命令读取文本文件
rating_data_raw = sc.textFile("/<path_to_csv_file>.csv")
是否有一种方法可以指定RDD rating_data_raw应该分割到的分区数？我想为更高的并发性指定大量的分区。
浏览 0提问于2015-12-14得票数 3
回答已采纳
1回答
我们有一个1.5BM记录分散在几个csv文件。为了生成一个count聚合，我们需要对几个列进行分组。
我们目前的战略是：
将它们加载到数据文件中(使用Dask或pyspark)
聚合列，以便生成2列作为键:value(我们不确定这是否值得)
将文件保存为Parquet
读取Parquet文件(Dask或pyspark)，并在dataframe的索引上运行groupby。
对于Parquet文件上高效的groupby来说，最佳实践是什么？
在索引上而不是在列(或一组列)上执行groupby有多大益处？
我们知道有一个partition可以提供帮助--但在我们的例子中，我们
浏览 2提问于2017-07-09得票数 3
我们有一个要求，从一个非分区的work_db.customer_tbl外部蜂窝表通过PySpark从一个分区的外部的蜂巢表通过PySpark摄取数据，以前是通过hive查询完成的。最后一个表由列 load_date (load_date列的格式为load_date)进行分区。
因此，我们有一个简单的PySpark脚本，它使用一个插入查询(与前面使用的单元查询相同)来使用spark.sql()命令摄取数据。但是，我们有一些严重的性能问题，因为我们在摄入后试图摄入的表大约有3000分区，和每个分区大约有4MB的数据，除了最后一个分区大约4GB。总表大小接近15 is。另外，在摄入每个分区之后，每
浏览 2提问于2021-03-25得票数 0
我有一个很大的csv文件(5 5GB)，其中包含大约50M行。我想让更小的块csv(~1M行)从原始的csv文件。我尝试了下面的方法来分块，但花了很多时间来执行分块：
1. I used Pandas to read data from csv and make chunk and write into chunk csv file.
import pandas
rows = pd.read_csv('test.csv', chunksize=1000000)
for i, chunck in enumerate(rows):
chunck.to_csv('
浏览 1提问于2020-09-02得票数 0
1回答
当在pyspark中使用partitionby()时，我应该采用什么方法来在一个文件夹而不是多个文件夹中编写csv文件？有什么建议吗？
码
from pyspark.sql import SparkSession
from pyspark import SparkConf
import pyodbc
appName = "PySpark Teradata Example"
master = "local"
conf = SparkConf() # create the configuration
conf.set("spark.repl.local.
浏览 5提问于2022-01-31得票数 0
2回答
假设我有一个带有20k行的csv文件，并将其导入。然后，我运行模型，如随机森林或Logistic回归从sklearn包，它运行良好。但是，当我导入PySpark数据格式并从PySpark包运行相同的模型(随机森林或Logistic回归)时，我会得到一个内存错误，我必须将csv的大小缩小到3-4k行。这一切为什么要发生？这是一个概念性的问题，还是我在某个地方编码错了？
对于Pandas dataframe，我的示例代码如下所示：
df=pd.read_csv("xx.csv")
features=TfIdf().fit(df['text'])
....
Rand
浏览 0提问于2019-02-06得票数 2
回答已采纳
我每天都将CSV文件存储在亚马逊网络服务s3中。下面是我的S3文件路径结构：
s3://data-dl/abc/d=2019-09-19/2019-09-19-data.csv
在此结构中，将每天生成s3文件路径的日期部分。
现在我想使用AWS glue for ETL将数据从S3传输到Redshift。要使用它，我如何在数据目录中添加S3路径？我只想同步最近使用的文件夹CSV文件。
另外，对于Job part，我如何在Glue Pyspark脚本中声明此动态路径？
浏览 1提问于2019-09-19得票数 0
我正在尝试使用Spark 2.0中的新东西将一些代码从Spark 1.6移植到Spark 2.0。首先，我想使用Spark 2.0的csv阅读器。顺便说一句，我用的是pyspark。
使用“旧的”textFile函数，我可以设置分区的最小数量。例如：
file= sc.textFile('/home/xpto/text.csv', minPartitions=10)
header = file.first() #extract header
data = file.filter(lambda x:x !=header) #csv without header
...
现在，使用
浏览 0提问于2016-07-01得票数 5
回答已采纳
使用PySpark，我有一些代码可以运行在一系列查询中。
for index, query in enumerate(query_map):
spark_dataframe.filter(
query).write.csv('s3://OutputBucket/Csvs/Query_{}'.format(index)
我是个新手，但我知道每个分区都在将各个csv文件写入一个名为Query_[index]的目录。现在，我想收集这些文件，并把它们放在熊猫的资料里。
import boto3
import pandas
s3 = boto3.resource(&
浏览 3提问于2018-02-07得票数 1
3回答
我正在尝试使用pyspark从databricks datalake中提升一些文件。为此，我使用"sqlContext“语句来创建数据框，这样做没有问题。每个文件都按创建日期命名，例如"20211001.cv“。这些信息每天都会出现，我使用"* .csv“来获取它们。但现在我需要将文件从某个日期向前移动，但我找不到方法，这就是为什么我会求助于您。 我使用的语句风格如下： df_example= (sqlContext
.read
.format("com.databricks.spark.csv")
浏览 58提问于2021-11-09得票数 1
df.write.option("header",True) \
.option("maxRecordsPerFile", 100000) \
.partitionBy("customer") \
.mode("overwrite") \
.csv("/mnt/Output/")
根据customer列，我正在使用上面的pyspark脚本将我的大CSV文件分割成小块。我正在使用Azure数据砖。
maxRecordsPerFile选项不起作用，它在每个
浏览 18提问于2022-10-12得票数 0
我有大约3500个csv，我将其转换为按日期划分的镶木地板(此数据跨越7天)。我想设置拼图文件的大小，使每个文件都是1 1gb。目前，我收到的文件太多(每天400-600个)，大小从64 MB到128 MB不等。我可以重新分区(使用repartition/coalesce)到x个文件，每个分区(天)，但我仍然有不同的文件大小，这取决于一天中存在的数据量，所以第一天可能有20 gb，所以10个文件每个2 gb，但第二天有10 gb，所以每个文件都是1 gb。我正在寻找如何设置/编码，使每个分区中的每个文件都是1 1gb。我正在使用pyspark，下面是我用来编写拼花面板文件的代码。
csv_re
浏览 2提问于2017-07-19得票数 0
我使用的是版本2.5.4中的pyspark函数。我在pyspark.sql.dataframe中有以下数据：
df = spark.createDataFrame(
[
(302, 'foo'), # values
(203, 'bar'),
(202, 'foo'),
(202, 'bar'),
(172, 'xxx'),
(172, 'yyy'),
],
['LU&#
浏览 0提问于2020-04-07得票数 0
当我保存一个用partitionBy分区的数据，然后读取它时，它将被读入多少个分区？我用下面的脚本来理解这种行为，但这对我来说毫无意义
我有一台八芯机器
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
import random
def f(num_key):
data1 = [(i,random.randint(1,5),random.randint(1,5)) for t in range(2) for i in range(num_key)]
df1=spark
浏览 4提问于2022-01-27得票数 0
1回答
我有大约500-700 csv文件的日志加载每周。文件名是每周随机生成的。这些文件存储在google云存储上，pyspark在google dataproc上运行，有几个工作节点。
我已经创建了一个pyspark文件，它基本上只是迭代文件并按原样加载到bigquery中。我以为我会在bigquery表上看到增量加载，但似乎pyspark会将整个gz输入文件合并到dataframe中，然后立即处理它。所以我在大约90分钟后得到了结果。
我如何编写pyspark代码，以便每次读取文件时，都会将其写入bigquery。总体时间可能不会有太大变化，但至少我如何才能看到部分写入bigquery的数据？
浏览 22提问于2021-10-10得票数 1
1回答
from pyspark.sql import SparkSession
spark= SparkSession.builder.master("local[4]").getOrCreate()
df = spark.read.csv("annual-enterprise-survey-2021-financial-year-provisional-size-bands-csv.csv")
df.createOrReplaceTempView("table")
sqldf = spark.sql('SELECT _c5 FROM tab
浏览 14提问于2022-07-21得票数 0
2回答
我使用pyspark来处理50 15的数据，使用15 m4的AWS。
数据的每一行都包含一天中特定时间的一些信息。我使用下面的for循环提取和聚合每小时的信息。最后，我union数据，因为我希望我的结果保存在一个 csv文件中。
# daily_df is a empty pyspark DataFrame
for hour in range(24):
hourly_df = df.filter(hourFilter("Time")).groupby("Animal").agg(mean("weights"), sum("is_m
浏览 0提问于2016-09-15得票数 4
回答已采纳
我有一个正在运行的Google Dataproc集群，并且正在向它提交一个PySpark作业，该作业从Google Cloud Storage (945MB CSV文件，400万行-->总共需要48秒才能读入)读取一个文件到PySpark dataframe，并对该数据帧应用一个函数(parsed_dataframe = raw_dataframe.rdd.map(parse_user_agents).toDF() -->大约需要4到5秒)。
然后，我必须将这些修改后的结果作为GZIP的CSV或Parquet文件保存回Google Cloud Storage。我还可以将这些修改后的
浏览 8提问于2017-08-01得票数 4
我有两个脚本:a和b。在脚本'a‘中，两个CSV文件被读取到两个数据帧中，然后连接到一个结果数据帧中，该数据帧随后被写入CSV文件。此任务不会出现OOM问题，并且速度非常快:10亿行、100列、41.2 GB CSV文件的处理时间为8-9分钟。 另一个脚本'b‘在每个方面都与'a’相似，除了一个方面:书写格式。输入文件是相同的: 1B行，100个cols，41.2 GB csv文件。此脚本将生成的数据框保存为ORC格式。然后它会导致一个错误： An error occurred while calling o91.orc. Job aborted due to sta
浏览 18提问于2019-03-13得票数 1
回答已采纳
这是这个线程的延续。
我正在努力将我的pyspark数据框架df保存在我的pyspark 3.0.1中。所以我写了
df.coalesce(1).write.csv('mypath/df.csv)
但是在执行这个命令之后，我在mypath中看到了一个名为df.csv的文件夹，它包含以下4个文件
1._committed_..
2._started_...
3._Success
4. part-00000-.. .csv
你能告诉我如何用df.csv保存所有数据吗？
浏览 6提问于2021-01-29得票数 4
回答已采纳
我有一个pyspark作业，它将结果数据帧写入本地文件系统中。目前它在local模式下运行，所以我执行coalesce(1)来获取单个文件，如下所示 file_format = 'avro' # will be dynamic and so it will be like avro, json, csv, etc
df.coalesce.write.format(file_format).save('file:///pyspark_data/output') 但我发现了很多内存问题(OOM)，而且也需要更长的时间。因此，我希望以yarn身份运行此作业
浏览 68提问于2020-08-26得票数 0
我正在尝试从一个分区文件夹中加载pyspark csv : mnt/data/test/ingestdatetime=20210208/test_20210208.csv df = spark.read.csv("mnt/data/test")
df = df.filter(df['ingestdatetime'] == '20210208') 基本上，我想看看模式是否与它应该的不同(数据没有标头，所以我不能比较标头) 问题是，每当我在顶层" data /test/“加载数据时，模式都是基于几行”推断“的，并且它不会看到新文件是否有额
浏览 16提问于2021-04-12得票数 0
回答已采纳
我有以下数据：
import pandas as pd
d = {'col1': [1, 2], 'col2': [3, 4], 'col3': [5, 6]}
df = pd.DataFrame(data=d)
我想按'col1‘对数据进行分区，但我不希望'col1’变量出现在最终数据中。这个是可能的吗？
下面的代码将按col1进行分区，但是如何确保'col1‘不会出现在最终数据中？
from pyspark.sql.functions import *
df.write.partitionBy("col1&#
浏览 3提问于2020-05-13得票数 0
我正在AWS EMR上运行一个5节点星火集群，每个大小为m3.xlarge (1个主站，4个从站)。我成功地运行了一个146 up bzip2压缩的CSV文件，最后得到了一个完美的聚合结果。
现在，我试图在这个集群上处理一个~5GB的bzip2 CSV文件，但是我收到了以下错误：
16/11/23 17:29:53警告TaskSetManager:在第6阶段(TID，xxx.compute.internal)中丢失的任务49.2 : ExecutorLostFailure (执行器16因运行任务之一而退出)原因:因超过记忆限制被纱线杀死的容器。使用10.4GB的10.4GB物理内存。考虑提高
浏览 8提问于2016-11-24得票数 64
我现在遇到了一些问题。
我有一大堆股票滴答数据csv文件。我需要计算每只股票的秒级数据。
我希望每个股票的数据可以在10秒内处理，合并成一个大文件，最后输出到csv。因为使用熊猫会受到我笔记本电脑记忆的限制，如果我想在熊猫身上做的话，我需要做很多读csv/ to _csv的工作。我认为这将占用时间，所以我选择了这样一种方式：(1)我使用pyspark读取所有的csv文件，生成一个大文件df。(2)我从df中获得了股票列表。然后进行迭代，每次我选择一个股票数据的pyspark，把它转移到熊猫数据中，在熊猫中计算它。最后，将该文件输出到本地文件中。
现在的问题是程序运行速度非常慢，对于一些股票，它
浏览 4提问于2020-04-19得票数 0
1回答
我如何让熊猫使用星火簇？
、、、
在pandas中的主要问题是它不能处理大量的操作数据，内存不足，无法处理大量的CSV文件，现在我在Hadoop中切换到Pyspark1.6，我已经尝试了dask.dataframe，但问题仍然存在，为什么让pandas与Hadoop集群或pyspark集群一起工作?我想在pandas中使用这个功能 import pandas as pd
df = pd.read_csv('text1.txt',names =['DATE','IMSI','WEBSITE','LINKUP','LINKDOWN',
浏览 12提问于2019-01-30得票数 3
回答已采纳
我使用"start_pyspark_shell“命令启动了我的spark shell，并给出了以下cli选项：-4个执行器，每个执行器2个核心，4 4GB内存用于工作节点，4 4GB用于主节点
存储: HDFS
输入文件:一个大小为221.3 MB的压缩.csv.gz文件(HDFS上为2个块)&
Spart版本: 2.4.0
手头的任务是计算文件中记录数的简单任务。唯一的问题是它是一个压缩文件。我使用以下命令加载了文件
df = spark.read.format("com.databricks.spark.csv").load(hdfs_path)
当我执行d
浏览 2提问于2020-09-01得票数 0
我目前正在做一个PySpark作业(Spark2.2.0)，它打算基于一组文档来训练一个潜在的Dirichlet分配模型。输入文档作为位于上的CSV文件提供。
下面的代码成功地运行在一个单一节点集群(4vCPU/ 15GB内存)上，其中包含一小部分文档(~6500)，生成的主题数量少(10)，迭代次数低(100)。但是，对于主题数量或迭代次数较大的文档集或较高值的其他尝试会迅速导致内存问题和作业失败。
另外，当将这个作业提交到一个4个节点集群时，我可以看到实际上只有一个工作节点在工作( CPU使用率为30%)，这让我认为代码没有为并行处理进行适当的优化。
代码
conf = pyspark.S
浏览 0提问于2017-08-25得票数 1
回答已采纳
3回答
我正在使用PySpark并加载一个csv文件。我有一个带有欧洲格式数字的列，这意味着逗号代替了点，反之亦然。
例如:我使用2.416,67而不是2,416.67。
My data in .csv file looks like this -
ID; Revenue
21; 2.645,45
23; 31.147,05
.
.
55; 1.009,11
在熊猫中，通过在decimal=','和thousands='.'选项中指定pd.read_csv()来读取欧洲格式，可以轻松地读取这样的文件。
Pandas代码：
import pa
浏览 0提问于2018-10-08得票数 13
回答已采纳
1回答
我正在使用内置Hadoop的PySpark 2.4.4处理一个大型数据集。数据集被分成多个压缩的CSV文件，我正在尝试使用PySpark的DataFrames报告各种列的聚合计数。在我要求PySpark使用以下命令将DataFrame写入磁盘之前，它一直处理得很好
df.write.csv("out.csv")
这会导致下面的警告淹没终端
WARN RowBasedKeyValueBatch: Calling spill() on RowBasedKeyValueBatch. Will not spill but return 0.
我尝试使用spark.executor.me
浏览 0提问于2020-02-23得票数 0
我在目录下有4个文件，其中一个文件缺少包含该列数据的一列
但是当我加载到spark DF中时，它没有添加第二列，也没有填充null
file1.csv
name| first|second|
female| raj| tarun|
file2.csv
name| first|second|
female| raj| tarun|
浏览 0提问于2018-04-27得票数 0
1回答
我有正在AWS Glue中爬行的数据。在那里，我使用PySpark并将其转换为拼图格式。我的原始数据是CSV，如下所示：
id, date, data 1, 202003, x 2, 202003, y 1, 202004, z等...
我能够成功地转换数据，但我不确定获得所需输出的最佳方法。在S3中，输出应按id和日期拆分。所以它应该有类似这样的东西：
s3://bucket/outputdata/{id}_{date}/{data}.parquet
其中id和date是数据中的实际id和日期值。文件的名称显然无关紧要，我只希望能够在S3对象前缀中创建“文件夹”并拆分其中的数据。
我是AWS
浏览 7提问于2020-06-06得票数 1
回答已采纳
这是我之前在上发布的一个问题的又一次跟进
我有以下数据集
其中有一个文件列表(约13万份)。在主目录中列出了它们的子目录，因此第一个单元可能是A/AAAAA，文件位于/data/A/AAA.csv。
这些文件都具有类似的格式，第一列称为日期，第二列是一系列都命名为值的列。因此，首先，值列名需要重命名为每个csv文件中的文件名。第二，帧需要完全外部连接，以日期为主要索引。第三，我希望保存文件并能够加载和操作它。文件大约在N行(日期数)X 130,001左右。
我正在尝试将所有文件完全连接到一个数据文件中，我以前尝试过使用熊猫，但是当我试图连接文件列表时内存不足，有人建议我尝试使用PySpark。
浏览 4提问于2020-02-18得票数 0
回答已采纳
3回答
在数据包中分裂csv文件
、、、、
我很新的火花，我仍然是我的第一次测试与它。我安装了一个节点，并将它作为我的主服务器运行在一个很好的服务器上：
pyspark --master local[20]
当然，我在使用火花放电的第一步中遇到了一些困难。
我有一个40 it的CSV文件和大约3亿行。我想要做的是找到最快的方法来分割这个文件，并制作它的小包，并将它们存储为CSV文件。为此，我有两种设想：
第一个。不需要任何条件就将文件拆分。只需将其平分成100块(每行300万行)。
第二个我正在加载的CSV数据是一个表格数据，我有一个列X，它有100 K不同的is。我想要做的是创建一组字典，并创建更小的CSV文件，在这些文件中，我的字典
浏览 0提问于2018-04-16得票数 0
回答已采纳
我是新来的PySpark (或火花在这件事上)。我将Python列表转换为RDD。
name_list_json = [ '{"name": "k"}', '{"name": "b"}', '{"name": "c"}' ]
name_list_rdd = spark.sparkContext.parallelize(name_list_json)
print(name_list_rdd)
这打印出"ParallelCollectionRDD2
浏览 0提问于2021-01-28得票数 0
回答已采纳
1回答
使用pyspark，我在HDFS中将不同的电子邮件列表从hive表写入到平面文件中。限制是我的文件不应该包含超过4.5M行。
我写了这段代码：
df = spark.read.table(working_table).select("email").distinct()
final_df = df.withColumn(
"rnk",
(F.row_number().over(Window.orderBy(F.lit(1))) / 4500000).cast("int")
)
final_df.groupBy("rn
浏览 9提问于2020-02-07得票数 1
这是我的pyspark代码： csv_file = "/FileStore/tables/mnt/training/departuredelays02.csv"
schema = "`date` STRING, `delay` INT, `distance` INT, `origin` STRING, `destination` STRING"
df = (spark
.read
.format("csv")
.option("header","t
浏览 23提问于2020-12-09得票数 0
1回答
我有一份简单的提交工作：
aggregateStores 读取CSV文件执行SQL聚合，并对进行排序，将结果显示为。
我将洗牌分区保持为4，可以看到总共创建了7个阶段(4个用于保存地板，2个用于显示操作，1个用于读取文件，但不确定)。
我有几个问题，，
为什么地板存储有4个阶段，不超过或少于这个阶段，以及与其他阶段相似的东西？
为什么Spark查询在后台启动阶段？
码
from pyspark.sql import SparkSession
from pyspark.sql.functions import lit
from pyspark.context import SparkC
浏览 6提问于2020-07-06得票数 2
回答已采纳
2回答
在pySpark中，我会在加载数据时更改分区的数量。
df_sp = spark.read\
.format('csv')\
.option("header", "true")\
.option("mode", "FAILFAST")\
.option("inferSchema", "true")\
.option("sep", "
浏览 2提问于2022-03-04得票数 -1
回答已采纳
1回答
Pyspark创建文件夹而不是文件。对于下面的命令，它在目录中创建一个名为proto.parquet的空文件夹。
df.write.parquet("output/proto.parquet")
尝试与csv和其他格式，但仍然一样。
浏览 0提问于2018-03-11得票数 0
嗨，我是Spark和Amazon EMR集群的新手。 我试图编写一个可以在Amazon EMR集群上运行的演示spark应用程序。当代码在Zeppelin notebook上运行时，它会返回输出，我认为输出会保存为Amazon EMR集群上的单个文件，如下所示： %pyspark
spark.conf.set('spark.sql.repl.eagerEval.enabled', True)
if type(df.c) == type(upper(df.c)) == type(df.c.isNull()):
df_new = df.withColumn('up
浏览 31提问于2021-04-16得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
