我在s3中有.dat文件。我需要通过spark读取该文件，并进行一些过滤，然后再次加载到S3中
在云计算领域中，S3是指Amazon Simple Storage Service，是一种对象存储服务，用于存储和检索大量数据。.dat文件是一种常见的数据文件格式，通常包含结构化数据。
要通过Spark读取S3中的.dat文件并进行过滤，可以按照以下步骤进行操作：
- 配置Spark环境：确保已经安装并配置好Spark集群，包括Spark的安装路径、环境变量等。
- 导入必要的库和模块：在Spark应用程序中导入所需的库和模块，例如pyspark、boto3等。
- 创建SparkSession：使用SparkSession对象来创建与Spark集群的连接，并设置相关的配置参数。
- 读取S3中的.dat文件：使用SparkSession的read方法，指定文件路径为S3中的.dat文件路径，读取文件内容。
- 读取S3中的.dat文件：使用SparkSession的read方法，指定文件路径为S3中的.dat文件路径，读取文件内容。
- 这里假设.dat文件是以CSV格式存储的，可以根据实际情况选择合适的文件格式和读取选项。
- 进行数据过滤：使用Spark的DataFrame API或SQL语句对读取的数据进行过滤操作，根据需要定义过滤条件。
- 进行数据过滤：使用Spark的DataFrame API或SQL语句对读取的数据进行过滤操作，根据需要定义过滤条件。
- 这里的column_name是要过滤的列名，filter_condition是过滤条件。
- 将过滤后的数据重新加载到S3中：使用Spark的write方法将过滤后的数据重新保存到S3中。
- 将过滤后的数据重新加载到S3中：使用Spark的write方法将过滤后的数据重新保存到S3中。
- 这里同样假设保存的文件格式为CSV，可以根据实际需求选择合适的文件格式和保存选项。
需要注意的是，上述代码中的"s3://bucket-name/path/to/file.dat"和"s3://bucket-name/path/to/filtered_file.dat"是示例路径，需要替换为实际的S3存储桶和文件路径。
推荐的腾讯云相关产品：腾讯云对象存储（COS），提供高可用、高可靠、低成本的对象存储服务，适用于大规模数据存储和访问场景。您可以通过腾讯云COS官方文档了解更多信息：腾讯云对象存储（COS）
请注意，以上答案仅供参考，实际操作可能因环境和需求而异。
相关·内容
1回答
我过去经常在本地运行Spark，并将文件分发给节点从来没有给我带来过问题，但是现在我正在将东西转移到Amazon集群服务中，并且开始出现故障。基本上，我正在使用Maxmind GeoLiteCity.dat来处理一些IP，我把它放在主服务器上的本地文件系统(file:///home/hadoop/GeoLiteCity.dat). )上。
下面是前面的一个问题，我使用了sc.addFile：
sc.addFile("file:///home/hadoop/GeoLiteCity.dat")
然后用这样的方式调用它：
val ipLookups = IpLookups(geoFi
浏览 2提问于2015-08-14得票数 3
回答已采纳
我有一个在S3位置有大约50列数据的.dat文件，需要通过spark读取，并为3列做一些过滤器，然后再次将数据加载到s3中
浏览 36提问于2019-04-13得票数 0
在从S3加载/读取拼图文件的情况下，spark 3.X是将完整的文件加载到内存中，然后执行过滤，还是像Hadoop一样加载过滤后的数据？
浏览 10提问于2021-03-11得票数 0
我有一个usecase，在s3中有多个文件，我是这样读的：
df1 = spark.read.csv("s3://bucket/fact/*.dat")
所有的.dat文件在开头都有6个数字，即PO id。
190234_purcahse.dat
125134_purcahse.dat
读取时，我需要在dataframe df1中作为一个新列来使用这个PO id。如何以最有效的方式实现这一点？在读取文件时，有任何方法获取文件名吗？
浏览 1提问于2021-11-16得票数 0
回答已采纳
我有一个EMR集群运行星火。在第一步中，CSV文件被转换为由paruqet.snappy列划分的date格式，因此我将使用
s3://my-bucket/dataset/date=2020-12-20/part-0001.parquet.snappy
s3://my-bucket/dataset/date=2020-12-20/part-0002.parquet.snappy
s3://my-bucket/dataset/date=2020-12-20/part-0003.parquet.snappy
s3://my-bucket/dataset/date=2020-12-20/part-00
浏览 6提问于2020-12-21得票数 1
回答已采纳
4回答
我正在将数据集加载到DynamicFrame中，执行转换，然后将其写回S3：
datasink = glueContext.write_dynamic_frame.from_options(
frame = someDateFrame,
connection_type = "s3",
connection_options = {
"path": "s3://the-bucket/some-data-set"
},
format = "parquet"
)
结果是12个拼
浏览 2提问于2019-10-06得票数 2
1回答
我正在开发一个关于Java的应用程序。生成并成功地将.jar加载到EMR集群。代码中有一行内容如下：
JsonReader jsonReader = new JsonReader(new FileReader("s3://naturgy-sabt-dev/QUERY/input.json"));
我百分之百肯定：
这样的档案确实存在。
在执行aws s3 cp s3://naturgy-sabt-dev/QUERY/input.json .时，我正在正确地接收.json文件。
设置IAM策略，以便绑定的EMR角色具有读取、写入和列表的权限。
不起作用。
浏览 0提问于2019-12-11得票数 0
回答已采纳
3回答
关于将s3上的文件读入带有Spark的EMR集群中的实际机制，只需问几个简单的问题：
spark.read.format("com.databricks.spark.csv").load("s3://my/dataset/").where($"state" === "WA")是否将整个数据集传送到EMR集群的本地HDFS，然后执行筛选？还是在将数据集带入群集时对记录进行筛选？还是两者都不起作用？如果是这样的话，到底发生了什么？
官方文档缺乏对正在发生的事情的解释(或者如果它有解释，我就找不到它)。有人能用这样的解释来解释，
浏览 0提问于2018-01-20得票数 1
回答已采纳
我们有很多在一年/月/日/小时内划分的拼花数据集。
其中有几个小时是空的，只有一个_SUCCESS文件。
我们实现了遍历所有分区并执行工作的作业。我们在打开空数据集时遇到了问题。与org.apache.spark.sql.AnalysisException: Unable to infer schema for Parquet. It must be specified manually.相撞
(使用EMR5.3.0 -我们正在努力摆脱依赖，以便升级到更高版本)
火花壳例子：
Welcome to
____ __
/ __/__ ___ ____
浏览 0提问于2019-04-02得票数 2
回答已采纳
我在亚马逊S3中有两个存储桶bucket1和bucket2。
aws s3 cp s3://bucket1 s3://bucket2 --recursive --exclude "*" --include "*am.dat"
不从bucket1复制任何文件，尽管bucket1具有匹配模式*am.dat的文件。
aws s3 cp s3://bucket1 s3://bucket2 --recursive
成功将所有文件复制到bucket2。
浏览 0提问于2018-11-05得票数 0
我创建了一个AWS EMR Hadoop集群，其中“AWS Glue Data Catalog”用于“for Spark table metadata”。因此，在Spark作业或spark-shell中，我可以编写使用Glue/Athena数据库和表的Spark SQL。
如果在EMR中运行的Spark作业正在读取该表的内容时更改了Athena表的位置，会发生什么情况？
假设我在Glue数据库中有一个名为"item“的雅典娜表，名为"my_db”。雅典娜表格位置指向存储包含数据的地块文件的S3文件夹。此文件夹为s3://my_bucket/item_2020_03_02。
在E
浏览 26提问于2020-03-03得票数 0
回答已采纳
我有大量的csv文件(大约20000个)，这些文件中的大多数大约有4,000列，10%的文件可以稍微少一些或多一些列。我想将这些文件从S3加载到spark中，从CSV文件中推断出模式，然后合并模式以处理混合模式文件。然后，在减少分区数量后，将其写回S3作为拼图。
val df = spark.read.format("csv").option("header", "true").option("mergeSchema", "true").option("inferSchema", "
浏览 1提问于2018-07-05得票数 1
2回答
我想知道在下面的场景中，数据是如何加载到spark中的：
有10 GB的交易数据存储在S3中的拼花格式，我将运行一个星火程序，以分类每一个记录在10 GB的parquet文件(例如收入，购物，晚餐)。
我有以下问题：
如何将这10 GB分配给星团中的不同工作人员？如果所有这些都发生在内存中，那么10 GB的文件是否加载到Spark中，然后将数据拆分并发送给执行程序？？如果其中一个执行器在作业运行期间崩溃，主程序会再次从S3加载10 GB文件并提取假定由崩溃的执行器处理的数据子集并发送给另一个执行器吗？
浏览 7提问于2021-02-16得票数 1
回答已采纳
1回答
Spark如何创建从S3读取的对象的分区？
我在S3中有一个118MB大小的对象。我将对象读入胶水动态框架。转换为spark数据帧并应用了一些转换。然后将数据写回S3。输出文件夹由大小为51MB和39MB的两个对象组成。Spark如何决定从S3读取的数据的分区方案？
代码如下：
df = glueContext.create_dynamic_frame.from_catalog(database = glueDatabase, table_name = glueTable).toDF()
df = df.filter('student.year != "2005"&#
浏览 1提问于2020-06-26得票数 1
2回答
我想从一个S3桶中读取所有的拼板文件，包括子目录中的所有文件(这些实际上是前缀)。
在S3 url中使用通配符(*)只适用于指定文件夹中的文件。例如，使用此代码将只读取target/文件夹下的拼花文件。
df = spark.read.parquet("s3://bucket/target/*.parquet")
df.show()
假设我在我的s3桶中有这样一个结构：
"s3://bucket/target/2020/01/01/some-file.parquet"
"s3://bucket/target/2020/01/02/some-file.pa
浏览 1提问于2020-10-08得票数 2
回答已采纳
我在AWS S3中存储了大量大气数据，我想用Apache来分析这些数据，但是加载到RDD中有很大的困难。我已经在网上找到了一些例子来帮助解决这个问题的不同方面：
-using h5py通过h5py.File(filename) ()读取本地存储的科学数据文件
-boto/boto 3以获取文本文件格式的数据，通过get_contents_as_string()从S3到Spark
通过-map将一组文本文件通过keys.flatMap(mapFunc)发送到RDD
但我似乎不能让这些部件一起工作。具体来说--如何从s3加载netcdf文件(使用boto或直接加载，而不是附加到使用boto)以便使
浏览 2提问于2017-04-04得票数 1
我们有一个传统的批处理应用程序，我们从多个来源(Oracle、Salesforce、FTP文件、Web日志等)获取数据。我们将传入的数据存储在S3存储桶中，并在EMR上运行Spark来处理数据，并加载到S3和Redshift上。
现在，我们正在考虑通过引入AWS Kinesis，然后使用电子病历中的Spark Structured Streaming来处理流数据并将其加载到S3和Redshift，从而使该应用程序接近实时。假设我们有不同种类的数据，例如来自甲骨文的100+表、100+ salesforce对象、来自FTP location的20+文件、网络日志等，那么在这里使用AWS Kine
浏览 20提问于2018-09-10得票数 1
我有一个文件夹，其中有多个json文件(first.json，second.json) .Using scala我正在加载所有的json文件数据到rdd/ spark的数据集，然后对数据应用过滤器。
这里的问题是，如果我们有600个数据，那么我们需要将所有数据加载到rdd/dataset中，然后应用筛选器
寻找一个解决方案，我可以过滤记录，同时从文件夹本身读取，而不是加载到火花内存。
过滤是基于blockheight属性进行的。
每个文件中的Json结构：
first.json：
{“IsFee”："TransactionHash":"0xe6fc01ff633b417
浏览 8提问于2019-11-08得票数 0
我在不同的服务器中有日志文件(5台服务器通过局域网连接)，我需要处理并获得结果
每个节点都有4TB日志文件，我使用HDFS将所有日志文件加载到Spark中
每次请求出现时，火花加载所有文件(5 * 4TB)，然后使用Spark查询
如果我将所有日志文件加载到caseesndra中，然后进行查询(可以预加载)，该怎么办？哪条路快..？
浏览 5提问于2016-02-16得票数 0
回答已采纳
1回答
根创建的文件未上载到AWS S3
、、、、
我在一个目录(MacOS)中有两个文件(都小于5KB)。第一个文件(file1.txt)是根用户创建的，具有-rw-------文件权限。第二个文件(file2.txt)是由具有文件权限-rw-r—-r—-的常规用户配置文件创建的。当我试图将file1.txt (根创建的文件)上传到S3时，AWS控制台会挂起而不会上传该文件。当我试图通过AWS将file1.txt上传到S3时，该命令发出错误消息“警告:跳过文件file1.txt文件/目录不可读”。当我试图通过控制台上传file2.txt (用户帐户创建的文件)时，该文件成功地上载到S3。
为什么根创建的文件不上传到S3？如何将此文件上载到S3
浏览 4提问于2022-12-01得票数 -1
回答已采纳
3回答
我有一个S3存储桶，其中包含多个文件，这些文件的文件名中有冒号。
示例：
s3://my_bucket/my_data/en/2015120/batch:222:111:00000.jl.gz
我正在尝试将其加载到spark RDD中并访问第一行，如下所示。
my_data = sc.textFile("s3://my_bucket/my_data/en/2015120/batch:222:111:00000.jl.gz")
my_data.take(1)
但这是个问题，
llegalArgumentException: java.net.URISyntaxException:
浏览 1提问于2015-12-05得票数 3
1回答
我们正在试验将数据从亚马逊S3加载到Spark2.3集群中，该集群是在Mesosphere DC/OS下配置的。当我们在spark shell上运行代码时，spark无法识别S3文件系统：
File "/root/spark/spark-2.3.0-bin-hadoop2.7/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache
浏览 50提问于2018-06-07得票数 0
回答已采纳
我有一个火花应用程序试图读取一个文件。由于Spark的延迟加载，文件可能在spark.read时存在，但当我实际加载文件(如count操作)时，文件就会被删除。
// t0: file exists when initially trying to load the file
val ds = spark.read.json("s3://some-location/some-file")
// some operations on ds
// t1: the file s3://some-location/some-file is deleted from S3 by so
浏览 6提问于2022-03-02得票数 -2
通常，spark文件保存在多个部分中，允许每个工人读取不同的文件。在处理单个文件时，是否有类似的解决方案？s3提供了应该允许这种行为的select API。
spark似乎支持这个接口()，但这似乎只与优化查询有关，而与并行读取无关
浏览 0提问于2019-07-07得票数 3
我已经编写了pyspark作业来加载s3存储桶中存在的文件。在s3中有太多的小文件，我在spark中一个接一个地读文件。我正在一个接一个地读取文件，因为我添加了一个列，该列具有文件所在的存储桶路径的值。由于这个spark作业花费了大量的时间，因为它忙于一个接一个地迭代文件。 下面是实现该功能的代码： for filepathins3 in awsfilepathlist:
data = spark.read.format("parquet").load(filepathins3) \
.withColumn("path_
浏览 7提问于2020-12-08得票数 0
1回答
目前，我正在使用以下架构。我确实有一个DocumentDB数据库，该数据库使用DMS (CDC任务)将数据导出到S3，一旦这些数据被加载到S3上，我需要将其加载到Databricks中。
我已经能够读取CSV内容(它有大量的JSONS)，但我不知道如何解析/插入到Databricks表中。
遵循导出到S3的JSON有效负载。
{
"_id": {
"$oid": "12332334"
},
"processed": false,
"col1": "000000
浏览 24提问于2022-03-21得票数 0
我已经将文件加载到azure存储帐户gen2中，并且正在使用Azure Synapse Analytics来查询它们。遵循这里的文档：，我应该能够创建一个spark sql表来查询分区数据，从而随后在我的sql on demand查询中使用来自spark sql的元数据来给定文档中的行：When a table is partitioned in Spark, files in storage are organized by folders. Serverless SQL pool will use partition metadata and only target relevant fo
浏览 20提问于2021-06-08得票数 1
我正在使用pyspark对csv文件做一些工作，因此我需要导入从下载的软件包-csv_2.10-1.4.0.jar
由于代理问题，我将jar下载到本地。
有人能告诉我引用本地jar的正确用法是什么吗？
下面是我使用的代码：
pyspark --jars /home/rx52019/data/spark-csv_2.10-1.4.0.jar
但是，当我运行时，它将像预期的那样将我带到pyspark shell：
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true'
浏览 3提问于2017-08-03得票数 0
1回答
自动从S3加载数据到红移
、、、
我希望将数据从S3加载到Redshift。以每秒大约5MB{近似大小}的速度进入S3的数据。
我需要自动将数据从S3加载到Redshift。S3的数据是从kafka流消费者应用程序中转储出来的。
文件夹S3数据在文件夹结构中。示例文件夹：bucketName/abc-event/2020/9/15/10
此文件夹中的文件：
abc-2020年-9-15-10-0-01-abc。5MB
abc-2020年-9-15-10-0-02-aasdljc。5MB
abc-2020年-9-15-10-0-03-thntsfv。5MB
S3中的文件将json对象与下一行分隔开。此数据需要加载到红移
浏览 9提问于2020-09-10得票数 1
回答已采纳
2回答
我正在尝试使用AWS EMR的Pyspark来读取它驻留在s3中的Excel文件，为此，我下载了spark-excel jars spark-excel_2.11-0.12.4.jar和spark-excel_2.12-0.13.5.jar，并将其放入s3存储桶中
scenario 1:
===========
df = spark.read.format("com.crealytics.spark.excel").option("useHeader", "true").option("inferschema", "tru
浏览 109提问于2021-02-13得票数 0
配置单元中有一个外部表指向未分区的s3位置。该表指向s3中的一个文件夹，但数据位于该文件夹内的多个子文件夹中。
即使没有对表进行分区，也可以通过在配置单元中设置一些属性来查询该表，如下所示：set hive.input.dir.recursive=true; set hive.mapred.supports.subdirectories=true; set hive.supports.subdirectories=true; set mapred.input.dir.recursive=true;
然而，当在spark中使用相同的表使用df = sqlContext.sql("sele
浏览 2提问于2017-10-16得票数 1
2回答
如何从Hive表直接创建RDD？
、、、、
我正在学习spark，并使用SparkContext对象和一些本地文件、s3和hdfs创建rdd，如下所示：
val lines = sc.textFile("file://../kv/mydata.log")
val lines = sc.textFile("s3n://../kv/mydata.log")
val lines = sc.textFile("hdfs://../kv/mydata.log")
现在我在Hive表中有了一些数据。是否可以直接加载配置单元表，并将该数据用作RDD？
浏览 0提问于2019-03-08得票数 1
2回答
我们的要求是使用S3将csv和parquet文件加载到数据文件中。
我使用的代码是：
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
conf = SparkConf()
appName = "S3"
master = "local"
conf.set('spark.executor.extraJavaOptions', '-Dcom.amazonaws.services.s3.enableV4=true')
con
浏览 25提问于2022-10-30得票数 -1
如何将S3存储桶中的大量文件加载到单个PySpark数据帧中？我在一个EMR实例上运行。如果文件是本地的，我可以使用SparkContext textFile方法。但是当文件在S3上时，我如何使用boto3加载多个不同类型的文件(CSV，JSON，...)转换成一个单独的数据帧进行处理？
浏览 0提问于2018-05-29得票数 3
回答已采纳
1回答
当我在本地使用spark，在本地文件系统上写入数据时，它会创建一些有用的.crc文件。
在Aws EMR和S3上使用相同的工作，.crc文件不会被写入。
这是正常的吗？有办法强制在.crc上写入S3文件吗？
浏览 1提问于2018-10-15得票数 0
回答已采纳
DataSet<Row> a = spark.read().format("com.memsql.spark.connector").option("query", "select * from a");
a = a.filter((row)-> row.x = row.y)
Sring xstring = "...select all values of x from a and make comma separated string"
DataSet<Row> b = spark.read().f
浏览 20提问于2020-02-15得票数 0
回答已采纳
独立程序可以通过使用AWS客户端jar文件来读写AWS S3文件而不需要Hadoop。火花程序可以在没有Hadoop的情况下读写文件。然而，Spark需要读写AWS S3文件的程序才能使用Hadoop。即使如此，Spark1.4和Hadoop2.6& 2.7中仍然存在关于S3缺少Hadoop类的运行时错误，即使设置了Hadoop目录。
是否有一种方法可以让星火程序通过使用AWS客户端jar文件来读取/写入S3文件而不使用Hadoop？
如果不是，我如何在运行时解决S3中缺少Hadoop类的Spark问题？
浏览 2提问于2015-07-26得票数 4
1回答
我在S3桶中有一些快速压缩的Parquet文件，我正在尝试读取这些文件的数据并打印它的模式。但是，当我从文件夹路径读取时，模式与单个文件的模式不同。
文件夹路径架构：
df = spark.read.parquet("s3://bucket_name/rds-aurora/core/")
df.printSchema()
root
|-- rid: long (nullable = true)
|-- id: string (nullable = true)
|-- revision: integer (nullable = true)
|-- type: intege
浏览 5提问于2021-02-10得票数 1
回答已采纳
我正在尝试使用R sparklyr将多个拼图文件加载到一个Spark表中。附加的代码显示了我是如何做到这一点的。
spark_load_data <- function(db_conn, test_period)
{
library(DBI)
#
overwrite <- TRUE
#
for (ts in seq(as.Date(test_period["START_DATE","VALUE"]),
as.Date(test_period["END_DATE
浏览 0提问于2019-08-17得票数 0
1回答
S3桶中的对象大小为5.3GB。为了将对象转换为数据，我使用了get_object("link to bucket path")。但这会导致记忆问题。
因此，我在RStudio中安装了Spark2.3.0，并试图将该对象直接加载到Spark中，但是直接将对象加载到Spark中的命令不为人所知。library(sparklyr) library(dplyr) sc <- spark_connect(master = "local")
如果我将对象转换为可读的数据类型(例如数据帧/tbl(R中的data.framework/tbl))，我将使用copy_to将
浏览 2提问于2018-07-30得票数 0
回答已采纳
我想从 (on )从加载数据。我注意到Databricks提供了很多对亚马逊S3的支持，但对谷歌却没有提供支持。
从Google BigQuery将数据加载到Spark (在数据库上)的最佳方法是什么？会允许我这样做吗?还是只对托管在Google存储上的文件有效？
浏览 13提问于2016-04-05得票数 2
回答已采纳
我正在尝试从S3中将大约一百万个压缩文本文件读入spark。每个文件的压缩大小在50 MB到80 MB之间。总共大约有6.5万亿字节的数据。
不幸的是，我遇到了一个内存不足的异常，我不知道如何解决。像这样简单的东西：
raw_file_list = subprocess.Popen("aws s3 ls --recursive s3://my-bucket/export/", shell=True, stdout=subprocess.PIPE).stdout.read().strip().split('\n')
cleaned_names = ["s3
浏览 20提问于2016-07-26得票数 4
回答已采纳
我有一个存储.gz文件(json格式)的S3存储桶。每小时将有更多的文件发送到此存储桶。我想使用Glue增量地(每天)从S3存储桶中读取数据，将.gz转换为拼接，然后写回另一个S3存储桶。 我想我可以使用Glue书签来完成增量文件的读取/转换/写入。然而，我发现如果我读取spark dataframe中的.gz文件，书签就不起作用了。换句话说，我的下面的胶水作业不会以增量方式读取文件。它读取该存储桶中的所有文件。我确实在胶水作业中启用了Bookmark。 我错过了什么吗？是否需要通过glue动态数据帧而不是spark数据帧读取文件？ 我实际上不知道如何通过glue动态数据帧正确读取.gz文件
浏览 24提问于2020-12-30得票数 1
我在S3中有一个CSV文件，它有420行和54000列。我正在尝试将CSV加载到火花DataFrame中：
genoExp = sqlContext.read.load("/mnt/%s/RNA-Seq/GSE10846_Gene_Expression_Data.csv" %MOUNT_NAME, format='com.databricks.spark.csv', header='true', inferSchema = 'true')
这将返回一个错误：
com.univocity.parsers.common.TextP
浏览 3提问于2017-02-09得票数 0
使用spark写一个文件s3通常会创建一个有11个文件成功的目录，另一个文件名以s3中有实际数据的部分名称开头，如何使用pandas dataframe加载相同的文件，因为文件路径会改变，因为所有10个有实际数据的文件在每次运行中解析的文件名都会不同。
例如，编写时的文件路径：
df.colaesce.(10).write.path("s3://testfolder.csv")
存储在目录中的文件包括：
- sucess
- part-00-*.parquet
我有一个python作业，它将文件读取到pandas dataframe。
pd.read(s3\\.........
浏览 0提问于2020-08-17得票数 0
1回答
我在亚马逊s3上使用pyspark读取s3存储桶上的对象。我的存储桶是由许多json文件组成的，我读取了这些文件，然后将其另存为拼图文件
spark.read.json('s3://my-bucket/directory1/')
spark.write.parquet('s3://bucket-with-parquet/', mode='append')
每天我都会上传一些新的文件到s3://my-bucket/directory1/上，我想更新它们到s3://bucket-with-parquet/，有没有一种方法可以确保我不会两次更新数据。我
浏览 1提问于2019-12-03得票数 2
1回答
我必须读取上传到s3上的压缩文件。
功能:当任何文件上传到s3时，都会触发一个lambda，这会触发一个spark作业。
我应该在哪里读取文件，在AWS Lambda中还是通过Apache Spark？哪一个将是有益的？如何在spark中读取压缩文件？
浏览 2提问于2019-06-08得票数 1
8回答
我在将多个文件从AWS S3桶下载到本地机器时遇到困难。
我有我想下载的所有文件名，我不想要其他文件名。我怎么能这么做？在aws中有什么循环，我可以做一些迭代吗？
我需要下载几百个文件，这样就不可能使用一个将所有文件名作为参数的命令。
浏览 17提问于2016-06-24得票数 56
回答已采纳
我有一个Spark 2.0.2集群，我正在通过Pyspark通过Jupyter Notebook访问它。我有多个以管道分隔的txt文件(加载到HDFS中。我需要使用spark-csv将其加载到三个独立的数据帧中，具体取决于文件的名称。
我认为我可以采取三种方法--或者我可以使用python以某种方式遍历HDFS目录(还没有想出如何做到这一点，加载每个文件，然后执行联合。
我还知道在spark中有一些通配符功能(参见) -我可能可以利用
最后，我可以使用pandas从磁盘加载vanilla csv文件作为pandas数据帧，然后创建spark数据帧。这里的缺点是这些文件很大，加载到单个节点上的内
浏览 0提问于2016-12-14得票数 7
回答已采纳
2回答
我在一个文件中有一个s3键列表，我希望将它们的值输入到一个RDD中。正确的方法是什么？
下面的代码是我当前的尝试。从s3中提取文件的调用在map函数之外工作，但会在其内部导致空指针异常。我相信这是因为您不能在地图调用中创建或操作RDDs，但是我不知道如何解决这个问题。是否有方法可以在不将文件放入RDD的情况下从S3读取文件？如果我可以将它们直接读入字符串或inputStream中，这将是理想的，但我不相信Source支持s3a格式，它必须是从hadoop文件系统读取的东西。
JavaRDD<String> keys = spark.sparkContext().textFile(&
浏览 4提问于2017-11-02得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
