如何录制流数据 S3?

文件上 S3 不是
https://forums.aws.amazon.com/ ... 40395
. "append" 意思是B. S3 使用包含其他数据的新对象替换现有对象。

你应该看看方法 mode 为了 dataframewriter 在
http://spark.apache.org/docs/1 ... .html
Spark :

public DataFrameWriter mode/SaveMode saveMode/

表示数据或表已存在时的行为。 选项
使能够： - SaveMode.Overwrite: 覆盖现有数据。 -

SaveMode.Append: 添加数据

. - SaveMode.Ignore: 忽视
手术 /它是。 no-op/. - SaveMode.ErrorIfExists: 默认选项
在执行期间发出异常。

你可以尝试类似的东西

Append

savemode.

rdd.toDF.write

 .format/"json"/

 .mode/SaveMode.Append/

 .saveAsTextFile/"s3://iiiii/ttttt.json"/;

https://spark.apache.org/docs/ ... ppend
:

添加模式意味着在保存时 DataFrame 到数据源IF
数据 / 该表已经存在，预计
, 那个内容 DataFrame 将添加到现有数据中。

原则上，您可以通过传递关键字选择要作为输出格式的格式 "format" 方法

public DataFrameWriter format/java.lang.String source/

指定输出的基本源。 内置选项包括 "parquet", "json" 等等。

例如，如

parquet

:

df.write//.format/"parquet"/.save/"yourfile.parquet"/

或者怎么样

json

:

df.write//.format/"json"/.save/"yourfile.json"/

编辑：添加了详细信息 S3 会计

数据：

安装凭据有两个不同的选项，我们可以看到它
https://github.com/apache/spar ... 23L82
具有可变环境

System.getenv/"AWS_ACCESS_KEY_ID"/

或者与财产

spark.hadoop.foo

:

SparkHadoopUtil.scala:

if /key.startsWith/"spark.hadoop."// {

 hadoopConf.set/key.substring/"spark.hadoop.".length/, value/

}

所以你需要得到

hadoopConfiguration

在
http://spark.apache.org/docs/l ... ion//
或者
http://spark.apache.org/docs/l ... ation
并安装

hadoopConfiguration.set/"fs.s3.awsAccessKeyId", myAccessKey/

hadoopConfiguration.set/"fs.s3.awsSecretAccessKey", mySecretKey/