通过spark dataframe读取S3文件时，粘合书签不起作用
是因为粘合书签（glue catalog）是AWS Glue的一项功能，用于管理和维护数据目录和元数据。然而，由于要求不能提及亚马逊AWS，因此无法使用粘合书签来解决该问题。
在这种情况下，可以考虑以下解决方案：
- 检查S3文件路径：确保提供的S3文件路径是正确的，并且具有正确的权限。可以使用AWS S3控制台或AWS CLI验证路径是否正确。
- 检查Spark配置：确保Spark配置中正确设置了S3访问凭证。可以通过设置以下配置参数来提供凭证信息：
- 检查Spark配置：确保Spark配置中正确设置了S3访问凭证。可以通过设置以下配置参数来提供凭证信息：
- 这些参数可以在Spark应用程序中通过
SparkSession对象的sparkContext属性进行设置。 - 检查Spark版本兼容性：确保使用的Spark版本与S3文件的格式兼容。某些Spark版本可能对S3文件格式有特定的要求，例如Parquet文件需要Spark 2.0及以上版本。
- 使用其他方式读取S3文件：如果以上方法仍然无法解决问题，可以尝试使用其他方式读取S3文件，例如使用AWS SDK for Java或Python编写自定义代码来读取S3文件。
需要注意的是，由于要求不能提及亚马逊AWS，无法提供腾讯云相关产品和产品介绍链接地址。建议在遇到问题时，参考相关云计算厂商的文档和社区支持来获取更详细的解决方案。
相关·内容
我有一个存储.gz文件(json格式)的S3存储桶。每小时将有更多的文件发送到此存储桶。我想使用Glue增量地(每天)从S3存储桶中读取数据，将.gz转换为拼接，然后写回另一个S3存储桶。 我想我可以使用Glue书签来完成增量文件的读取/转换/写入。然而，我发现如果我读取spark dataframe中的.gz文件，书签就不起作用了。换句话说，我的下面的胶水作业不会以增量方式读取文件。它读取该存储桶中的所有文件。我确实在胶水作业中启用了Bookmark。 我错过了什么吗？是否需要通过glue动态数据帧而不是spark数据帧读取文件？ 我实际上不知道如何通过glue动态数据帧正确读取.gz文件
浏览 24提问于2020-12-30得票数 1
2回答
我在S3中有数据，我可以在S3中加载数据，因为RDD应用一些更改将其转换为dataframe并运行spark-sql查询。但是，每当向S3添加新数据时，我都需要加载整个数据，因为RDD会将其转换为dataframe并运行查询。有没有一种方法可以避免加载整个数据而只加载新数据。是否应该将新数据添加到RDD中，而不是加载整个RDD？
浏览 10提问于2019-04-30得票数 0
回答已采纳
2回答
我可以写入拼图格式，并按如下所示的列进行分区：
jobname = args['JOB_NAME']
#header is a spark DataFrame
header.repartition(1).write.parquet('s3://bucket/aws-glue/{}/header/'.format(jobname), 'append', partitionBy='date')
但是我不能用Glue的DynamicFrame做到这一点。
header_tmp = DynamicFrame.fromDF(header,
浏览 2提问于2018-03-07得票数 11
回答已采纳
我在火星雨方面还很新，我还在努力弄清楚到底是谁干的。我想要做的是在内存中加载一个拼花文件之后，使用pyarrow，我试着使它成为。但我搞错了。
--我要提到的是，我并不是直接阅读pyspark，因为s3中的文件给了我另一个关于“s3"的错误，所以我试图解决这个问题。下面我有一个可重复的例子。
import pyarrow.parquet as pq
import s3fs
s3 = s3fs.S3FileSystem()
parquet_file=pq.ParquetDataset('s3filepath.parquet',filesystem=s3)
sc = S
浏览 1提问于2019-07-29得票数 0
2回答
我是星火框架的新手，在我的本地机器上做一些小任务来练习。我的任务是:我在S3中存储了365个压缩的csv文件，其中包含每天的日志。我想要建一个全年的数据集。我的方法是从桶中检索密钥，构建每日数据格式，将它们统一为月份数据，对它们进行同样的操作，并作为回报获得全年数据。
它适用于我检索到的用于测试的一些样本数据。在构建DataFrames之前，我对文件进行解压缩，将未压缩的csv文件写入磁盘，并使用它创建DataFrame。
问题是:如果我从磁盘中删除csv文件(使其成为临时文件)，在创建dataframe之后，我无法对dataframe执行任何操作(例如year_df.count())。抛出S
浏览 0提问于2017-01-09得票数 1
回答已采纳
我有一个非常大的每日gzipped文件序列。我正在尝试使用PySpark重新保存所有S3格式的文件，以供以后使用。
如果对于单个文件(例如，2012-06-01)，我这样做：
dataframe = spark.read.csv('s3://mybucket/input/20120601.gz', schema=my_schema, header=True)
dataframe.write.parquet('s3://mybucket/output/20120601')
它可以工作，但是因为gzip是不可拆分的，所以它运行在单个主机上，并且我没有从使用集群中获得
浏览 2提问于2017-12-16得票数 0
由于我们的方案是常量的，所以我们使用的是spark.read()，这比在s3中存储数据时从选项创建动态框架要快得多
所以现在想要使用动态框架从胶水目录中读取数据需要很多时间，所以想要使用spark Dataframe.read.format(“).option(”url“，"").option("dtable"，schema.table名称).load()
需要输入什么格式和url选项以及任何其他内容？？
浏览 1提问于2021-04-12得票数 3
使用spark编写文件s3通常会创建两个文件成功的目录，另一个文件名以s3中具有实际数据的部分名称开头，如何使用pandas dataframe加载相同的文件，因为文件路径会发生变化，因为文件名与实际数据在每次运行中都不同。 例如，编写时的文件路径: df。写。\testfolder.csv(“s3”colaesce) 存储在目录中的文件是成功的part-00- 我有一个python作业，它将文件读取到pandas dataframe。 Pd.read(s3.what是此处指定的路径.)
浏览 8提问于2020-07-26得票数 0
我的script失败了，因为在脚本运行时，用于绘制df的S3桶会被更新为新文件。我不在乎新来的文件，但很明显斯派克在乎。
我尝试在错误消息中添加REFRESH命令，但这不起作用，因为在执行时不可能知道新文件何时到达，因此无法知道将该命令放在何处。我尝试将该刷新命令放在脚本中的4个不同位置(换句话说，在脚本的不同位置调用它4次)--所有这些都带有相同的失败消息。
Caused by: java.io.FileNotFoundException: No such file or directory '<snipped for posting>.snappy.parquet
浏览 0提问于2018-11-14得票数 2
回答已采纳
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
回答已采纳
使用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
有办法从DataFrame的tempDir转储中创建RedShift吗？
我的用例是当作业失败时，我想重试，但是继续从转储到S3的临时数据转储，而不是再次从RedShift中重新获取数据集，这是非常大的！
加载代码执行以下操作：
val df1 = spark.read
.format("com.databricks.spark.redshift")
.option("url", jdbcUrl)
.option("dbtable", spmeTable)
.option("tempdir", tempDir)
浏览 1提问于2019-08-16得票数 1
3回答
我需要从我的S3桶里读150遍
df1 = spark.read.json('s3://mybucket/f1')
df2 = spark.read.json('s3://mybucket/f2')
...
df150 = spark.read.json('s3://mybucket/f150')
如何使这一过程自动化？
spark.read.json produces Spark Dataframe.
如果我试试奥斯卡的建议
import spark
your_dfs_list = [spark.read.json("s3://cw-m
浏览 13提问于2020-07-14得票数 1
回答已采纳
1回答
我必须读取上传到s3上的压缩文件。
功能:当任何文件上传到s3时，都会触发一个lambda，这会触发一个spark作业。
我应该在哪里读取文件，在AWS Lambda中还是通过Apache Spark？哪一个将是有益的？如何在spark中读取压缩文件？
浏览 2提问于2019-06-08得票数 1
我在PySpark中读取AWS Glue作业的数据时遇到问题：
数据从AWS消防软管(示例数据)发送到s3桶，存储为JSON，并使用snappy进行压缩。
我能够用spark.read.json()从遗留的spark.read.json读取数据，但是使用from_catalog或from_options方法无法使用Glue动态框架(模式根本没有解析)：
火花遗物DataFrame
# import from legacy spark read
spark_df = spark.read.json("s3://my-bucket/sample-json-hadoop-snappy/
浏览 1提问于2020-06-11得票数 4
回答已采纳
我需要用新数据对旧数据执行update insert (Upsert)。
伪代码：
old_data = spark.read.parquet('s3://bucket/old_data/')
new_data= spark.read.parquet('s3://bucket/new_data/')
common_records = old_data.join(new_data,on=opk,how="inner")
non_match_records = old_data.join(new_data,on=opk,how="left_
浏览 0提问于2021-09-24得票数 1
我正在尝试用Scala或python自动化我的spark代码，这就是我想要做的
s3存储桶中的文件格式为filename_2016_02_01.csv.gz
从s3存储桶中，spark代码应该能够选择文件名并创建数据帧
example Dataframe=sqlContext.read.format("com.databricks.spark.csv").options(header="true").options(delimiter=",").options(inferSchema="true").load("s3://
浏览 0提问于2016-02-04得票数 1
环境
Scala
Apache Spark: 2.2.1
AWS上的EMR : emr-5.12.1
内容
我有一个大的DataFrame，如下所示：
val df = spark.read.option("basePath", "s3://some_bucket/").json("s3://some_bucket/group_id=*/")
在s3://some_bucket上有JSON文件~1TB，它包含5000个group_id分区。我希望使用SparkSQL执行转换，并且每个group_id都不同。
星条旗代码如下：
浏览 0提问于2018-04-27得票数 0
回答已采纳
1回答
使用Spark时，我可以在前缀中使用*从多个存储桶中读取数据。例如，我的文件夹结构如下： s3://bucket/folder/computation_date=2020-11-01/
s3://bucket/folder/computation_date=2020-11-02/
s3://bucket/folder/computation_date=2020-11-03/
etc. 使用PySpark，如果我想读取11个月的所有数据，我可以这样做： input_bucket = [MY-BUCKET]
input_prefix = [MY-FOLDER/computation_date=2
浏览 20提问于2020-12-01得票数 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
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
回答已采纳
我正在尝试将拼板数据从另一个s3桶复制到我的s3桶中。我希望将每个分区的大小限制为128 MB。我认为默认情况下，spark.sql.files.maxPartitionBytes应该设置为128 MB，但是当我在复制后查看s3中的分区文件时，我会看到大约226 MB的单个分区文件。我看了这篇文章，它建议我设置这个星火配置键，以限制分区的最大大小：，但它似乎不起作用吗？
这是配置键的定义：
读取文件时要打包到单个分区的最大字节数。此配置仅在使用Parquet、JSON和ORC等基于文件的源时才有效。
我也有点困惑，这是如何与大小的书面拼花文件。
作为参考，我正在胶水1.0版spark 2.4
浏览 54提问于2020-06-30得票数 1
我有一个星火作业，它读取一些S3上的CSV文件，处理并保存结果，因为files.These CSV包含日语文本。
当我在本地运行此作业时，读取S3 CSV文件并将其写入本地文件夹中，日文字母看起来很好。
但是当我在我的星体集群上运行这个程序时，读取相同的S3 CSV文件并将拼图写到HDFS时，所有的日文字母都会被混淆。
运行在火花集群(数据被混淆)上
spark-submit --master spark://spark-master-stg:7077 \
--conf spark.sql.session.timeZone=UTC \
--conf spark.driver.extraJava
浏览 2提问于2020-05-18得票数 0
1回答
我有一个相对庞大的前提表(约15亿行)，我正试图使用AWS以拼花格式将它拉到AWS S3中。我使用spark读取表并将其写入S3。问题是，我不能一次从源表中提取所有数据，因为源DB将耗尽内存并发出抱怨。为了解决这个问题，我使用谓词选项并行地按下过滤器，这可以很好地提取2亿左右的数据块。但是，当我试图将这个数据写入S3时，需要将近半个小时才能完成：
df = spark.read.jdbc(url=host_url,
table="TABLENAME",
predicates=pre
浏览 0提问于2020-06-05得票数 2
回答已采纳
我创建了一个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
回答已采纳
2回答
我使用Spark2.1.0、Hadoop2.7.3、Hive2.1.1、Sqoop1.4.6和Ganglia 3.7.2 (从s3加载数据)在EMR ( 5.5.1 )上运行吡火花作业。有多个桶包含输入文件，所以我有一个函数，它使用boto遍历它们，并根据某种模式过滤掉它们。
集群大小：主=> r4.xlarge，Worker => 3 x r4.4xlarge
问题：函数getFilePaths返回一个s3路径列表，该列表直接提供给spark方法。
使用Dataframe
file_list = getFilePaths() # ['s3://some_bucket/lo
浏览 1提问于2018-06-13得票数 1
2回答
Kinesis将文件的持久性(在本例中为时间序列JSON )管理为一个文件夹层次结构，该层次结构由YYYY/MM/DD/HH (直到24小时编号)...great划分。
那么我如何使用Spark2.0来读取这些嵌套的子文件夹，并从所有的叶json文件中创建一个静态的Dataframe呢？数据阅读器有“选项”吗？
我的下一个目标是成为一个流DF，火龙将新文件持久化到s3中，使用Spark2.0中的新结构化流自然成为流数据的一部分。我知道这都是实验性的--希望有人以前曾使用过S3作为流文件源，在这里，数据会像上面所描述的那样被解析为文件夹。当然，更喜欢的是一个直接的动态流，但这个连接器上没有日期为
浏览 4提问于2016-10-30得票数 8
过度简化的场景：--在s3文件中生成每月数据的过程。每个月运行的字段数可能不同。基于s3中的这些数据，我们将数据加载到一个表中，并且我们手动地(在每次运行中字段的数目可能会随着添加或删除几个列而改变)运行一个metrics.There，因为很少的metrics.There会对这些数据进行更多的计算/转换，但是为了让初学者我呈现更简单的usecase版本。
Approach：考虑到模式少的特性，由于每次运行时s3文件中的字段数可能不同，只需添加/删除几个字段，这就需要在SQL中每一次进行手动更改，因此我计划探索Spark/Scala，这样我们就可以直接从s3中读取数据，并根据字段动态生成SQL。
浏览 7提问于2020-03-19得票数 1
我正在调试create_dynamic_frame.from_catalog不返回数据的问题，尽管我可以通过雅典娜查看数据。
Data指向S3文件夹，并且有多个具有相同结构的文件。文件类型为csv，分隔符为空格“"，由两列(string和json字符串)组成，没有标头。
这是CSV格式文件。
这是使用爬虫生成的雅典娜查询。
调试时没有从dataframe返回的结果，有什么想法吗？
浏览 2提问于2021-02-02得票数 2
1回答
我正在尝试使用spark从亚马逊s3获取一个文件(以DataFrame或RDD的形式)，执行一些简单的转换，然后将文件发送到DynamoDB上的一个表中。
在阅读了其他一些论坛帖子后，我开始了解到读/写DynamoDB需要使用hadoopRDD -这与spark中的RDD不同-也不同于我检索s3文件的方式。
我该如何将DataFrame/RDD从s3中的文件更改为hadoopRDD，这样我才能将其发送回来？
我正在使用scala并测试spark-shell中的所有东西。
再次提前感谢！
浏览 0提问于2016-05-26得票数 7
1回答
我是Spark的新手，但我正在尽可能多地阅读。我有一个小项目，其中多个数据文件(在gzip中)将每小时连续地放在一个S3存储桶中。我需要能够打开/读取这些gzip文件，并在它们之间合并/聚合数据。因此，我需要以整体的方式来看待它们。如何使用Amazon AWS提供的哪些技术和工具？我是在S3文件夹中创建临时文件，还是在内存中保留数据帧，或者使用某个数据库并在每小时后清除数据？所以，我寻找的不仅仅是一段代码，而是一些想法。
到目前为止，在亚马逊网络服务中，我已经编写了一个pyspark脚本，它一次读取一个文件，并在output S3文件夹中创建一个输出文件。但这给我留下了每个小时的多个输出文件。
浏览 0提问于2020-01-23得票数 0
1回答
我有s3中的源数据，我的spark/scala应用程序将在一个新的partition_id列上对它进行分区之后，读取这些数据并将其写成拼花文件。partition_id的值将通过从具有字母数字字符串值的另一个id列中获取前两个字符来导出。例如：
id = 2dedfdg34h, partition_id = 2d
将数据写入s3后，将为每个分区创建单独的分区文件夹，并且一切看起来都很好。例如：
PRE partition_id=2d/
PRE partition_id=01/
PRE partition_id=0e/
PRE partition_id=fg/
PRE partition_id=
浏览 1提问于2019-09-25得票数 1
回答已采纳
有没有办法用Spark直接读取.Z (大写)文件扩展名？
我知道带有spark的Scala可以直接读取gzip文件(.gz)，但当我尝试将压缩的Z文件(.Z)加载到Dataframe中时，它不起作用。
浏览 0提问于2020-05-06得票数 2
我已经通过spark dataframe创建了CSV文件，这些文件会自动加密KMS。
作为参考，我给出了一个创建这些KMS加密文件的示例代码片段。如果你在写的时候看到，我不会给任何KMS密钥。如果你说出根本原因，那将会很有帮助。
val df=spark.read.format("csv").option("header", "true").load("s3:///test/App_IP.csv")
df.createOrReplaceTempView("test")
val df1=spark.sql("
浏览 2提问于2021-01-26得票数 1
1回答
我是星火公司的新手，我正尽我所能去阅读和研究。目前我被困在这上面，我花了几天时间来解决这个问题。我已经成功地在3台机器(1台主机，2台奴隶)上建立了一个星火集群，并运行了一些示例。现在，我正在编写一个Python应用程序，它将读取csv文件，然后将JSON文件中的每一行拆分，并将它们全部上传到S3。以下是我的问题：
我已经将csv转换为Spark，使用DataFrame，如何将这个DataFrame分割成多个行并转换为JSON？我读过Spark函数，但是它应用于整个DataFrame，所以我如何在DataFrame的每一行上使用这个函数而不是整个DataFrame呢？
如何在我的应用程
浏览 5提问于2016-12-05得票数 0
回答已采纳
1回答
当我从s3桶将数据加载到pyspark中时，进行一些操作(连接、联合)，然后尝试覆盖前面读取的相同路径(' data /csv/')。我得到了一个错误：
py4j.protocol.Py4JJavaError: An error occurred while calling o4635.save.
: org.apache.spark.SparkException: Job aborted.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.sca
浏览 0提问于2019-06-12得票数 0
4回答
我一直在尝试使用Glue，但还没有让它在现有的S3存储桶中成功地创建一个新表。作业将在没有错误的情况下执行，但在S3中永远不会有任何输出。
下面是自动生成的代码：
glueContext.write_dynamic_frame.from_options(frame = applymapping1,
connection_type = "s3", connection_options = {"path":
"s3://glueoutput/output/"}, format = "json", transformation_ct
浏览 1提问于2017-09-21得票数 6
我知道我可以使用spark.read.csv()或spark.read.json()将CSV文件或JSON文件作为一个DataFrame加载，但是如果我的源文件实际上是不同格式的，例如，我的一些原始数据在CSV文件中，而另一些在JSON文件中呢？我是否可以创建一个包含来自CSV和JSON文件的数据的DataFrame？也许我应该分别加载它们并在DataFrames上使用JOIN操作，或者我应该首先将所有JSON数据转换成CSV并使用单个spark.read.csv()加载它们，但是我想听听您的建议，因为我对Spark还是新手。
浏览 0提问于2021-08-14得票数 0
回答已采纳
1回答
我们在单个AWS EC2实例上以本地模式运行Spark，使用
"local[*]"
然而，使用New工具和一个简单的“top”进行分析表明，我们16台核心机器中只有一个CPU核心用于我们编写的三个不同的Java火花作业(我们也尝试过不同的AWS实例，但只有一个内核曾经被使用过)。
Runtime.getRuntime().availableProcessors()报告16个处理器，sparkContext.defaultParallelism()报告16个。
我看过各种Stackoverflow本地模式问题，但似乎没有一个解决了这个问题。
任何建议都很感激。
谢谢
编辑:过程
浏览 2提问于2016-10-31得票数 6
回答已采纳
我正在尝试使用Spark Structured Streaming从Kafka topic中读取数据。Kafka Broker启用了SSL。因此，我需要将私有CA证书安装/导入到spark驱动程序和执行器上的TrustStore文件中。
我不能在主spark submit命令之前使用单独的步骤导入证书，因为spark脚本是动态提交的(从s3下载)。s3中的这个spark脚本包含有关私有CA证书文件(.pem)所在位置(在单独的s3位置上)的信息。
我查找了一些方法来做到这一点。大多数解决方案都需要创建RDD或DataFrame，并在其上调用Map或MapPartition函数(本质上定义分区)
浏览 6提问于2020-07-12得票数 0
我想运行胶水作业来完成从s3到Postgres的许多csv文件的ETL过程。每天都会将新文件写入s3源桶。当我运行那些文件的爬虫来生成带有模式的表时，而不是胶数据目录中的一个表，我会得到许多表，这意味着爬虫不会识别这些文件的模式是相同的。也许是因为有很多文件只有头文件，但没有内容。
因此，当我用向导创建胶水作业时，当被问到要使用哪个表时，我只从glue数据目录(基于最大的csv文件创建)中选择一个表。因此，在DB中，我只有来自最大文件的数据，而不是来自所有csv文件的数据。我想这是因为爬虫在胶水数据目录中创建那些表的同时，也保存了对应于这个表的文件列表，我在s3:/aws-glue-tempo
浏览 0提问于2020-03-26得票数 3
1回答
我正试图使用Databricks DataFrame包和，从EMR集群上的S3上的CSV源创建一个
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true').load('s3n://h2o-airlines-unpacked/allyears.csv')
df.first()
这不会在4m3.xlarge的集群上终止。我
浏览 0提问于2015-08-28得票数 3
我们每天上午11点在s3桶中收到1份csv文件。我在上午11:30使用Glue将此文件转换为拼花格式。
我已经启用了作业书签来不处理已经处理的文件。尽管如此，我看到一些文件正在被重新处理，从而创建了重复的文件。
我读了这些问题和答案，和
他们对工作书签有很好的理解，但仍然没有解决这个问题。
AWS文档说，它支持用于书签的CSV文件。
想知道是否有人帮助我理解什么是问题，如果可能的话，也是这样的解决方案:)
编辑：
按Prabhakar的要求在这里粘贴示例代码。
staging_database_name = "my-glue-db"
s3_target_path = "
浏览 1提问于2019-06-24得票数 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
回答已采纳
我已经创建了一个管道，其中数据摄取发生在Redshift和S3之间。我能够使用以下方法完成完整的加载： def readFromRedShift(spark: SparkSession, schema, tablename):
table = str(schema) + str(".") + str(tablename)
(url, Properties, host, port, db) = con.getConnection("REDSHIFT")
df = spark.read.jdbc(url=url, t
浏览 24提问于2021-09-28得票数 0
我正在用Scala编写一个ETL (2.4)作业，在;上读取带有glob模式的-separated CSV文件。数据被加载到一个DataFrame中，并包含一个列(假设它被命名为custom)，其中包含一个JSON格式的字符串(、多层嵌套)。其目标是从该列自动推断模式，以便可以为S3中Parquet文件上的写接收器构造模式。
这篇文章()建议，Spark2.4中的schema_of_json可以从JSON格式的列或字符串中推断模式。
以下是我尝试过的：
val jsonSchema: String = df.select(schema_of_json(col("custom")
浏览 0提问于2019-02-14得票数 9
我在用
AWS DMS从Oracle中提取数据
它降落到S3原木桶里
使用AWS Glue，我想编写pyspark代码，而不使用databricks产品将CDC数据与初始加载合并。
为了创建Delta表，我需要在spark上下文中专门导入哪些库？
我在“安全配置、脚本库和作业参数(可选)”下将delta-core_2.12-0.7.0.jar添加到依赖于Glue的路径中，在错误下面得到错误
文件“Script2020-11-08-19-29.39.py”，第54行，在spark.read.parquet('s3://rawbucket/.../fullload/..
浏览 3提问于2020-11-08得票数 0
5回答
AWS胶水书签
、、、
如何验证我的书签是否正常工作？我发现，当我在上一次完成后立即运行一个作业时，似乎仍然需要很长时间。为什么会这样呢？我以为它不会读取它已经处理过的文件？脚本如下所示：
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
## @params: [JOB_N
浏览 0提问于2018-12-11得票数 5
我在一个xml文件中有2700万条记录，我想把它推到下面的elasticsearch索引中，这是用scala编写的代码片段，我将创建一个星火作业jar并在AWS EMR上运行
我如何有效地利用火花来完成这个练习？请指点。
我有一个12.5GB的压缩xml，我正在将它加载到中。我是星火公司的新手.(我应该分割这个gzip文件吗？或者火花执行者会处理它？)
class ReadFromXML {
def createXMLDF(): DataFrame = {
val spark: SparkSession = SparkUtils.getSparkInstance("Spa
浏览 5提问于2020-08-20得票数 4
回答已采纳
我试图在亚马逊S3中将一个Spark DataFrame (超过20G)保存到一个json文件中，我保存数据帧的代码如下所示：
dataframe.repartition(1).save("s3n://mybucket/testfile","json")
但是我从S3得到了一个错误“你建议的上传超过了最大允许的大小”，我知道亚马逊允许的最大文件大小是5 5GB。
可以和Spark一起使用S3分块上传吗？或者有另一种方法解决这个问题？
顺便说一句，我需要的数据在一个单一的文件，因为另一个用户将下载它之后。
*我在使用spark-ec2脚本创建的3节点集群中使用ap
浏览 0提问于2015-04-28得票数 19
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
