flink（非EMR集群） 怎么将流式数据写入到oss ,有相关sink吗？

"https://help.aliyun.com/document_detail/423511.html?s此回答整理自钉群“【③群】Apache Flink China社区” cm=20140722.H_423511._.ID_423511-OR_rec-V_1"

对于 Flink 非 EMR 集群，如果需要将流式数据写入到阿里云 OSS 中，可以使用 flink-oss-connectors 提供的 OSS Sink。

flink-oss-connectors 是 Flink 官方提供的阿里云 OSS 连接器，可以方便地将 Flink 的数据流写入到阿里云 OSS 中。使用该连接器，需要先在 Maven 中添加 flink-oss-connectors 依赖，然后在 Flink 程序中引入 OSS Sink，并进行相应的配置，如指定 OSS 存储桶、设置 OSS 访问凭证等。

下面是一个使用 flink-oss-connectors 将数据流写入到阿里云 OSS 的示例代码：

import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.connectors.fs.bucketing.DateTimeBucketer
import org.apache.flink.streaming.connectors.fs.bucketing.OnCheckpointRollingPolicy
import org.apache.flink.streaming.connectors.fs.{SequenceFileWriter, StringWriter}
import org.apache.flink.streaming.connectors.oss.OSSConfig
import org.apache.flink.streaming.connectors.oss.OSSFileSystem
import org.apache.flink.streaming.connectors.oss.OSSFileSystemConfigKeys
import org.apache.flink.streaming.connectors.oss.writer.OSSWriterConfig
import org.apache.flink.streaming.connectors.oss.writer.OSSWriterFactory

val ossConf = new OSSConfig()
ossConf.setString(OSSFileSystemConfigKeys.ENDPOINT, "your-oss-endpoint")
ossConf.setString(OSSFileSystemConfigKeys.ACCESS_KEY_ID, "your-access-key-id")
ossConf.setString(OSSFileSystemConfigKeys.ACCESS_KEY_SECRET, "your-access-key-secret")

val ossSink = StreamingFileSink.forBulkFormat(new Path("oss://your-bucket/your-path"), OSSWriterFactory.builder[String]()
  .withOSSConfig(ossConf)
  .withBucket("your-bucket")
  .withWriterConfig(OSSWriterConfig.builder().build())
  .withKeyFunction(e => e._1)
  .build())
  .withBucketAssigner(new DateTimeBucketer("yyyy-MM-dd--HHmm"))
  .withRollingPolicy(OnCheckpointRollingPolicy.build())
  .build()

stream.addSink(ossSink)

在上述代码中，我们首先创建了一个 OSSConfig 对象，并设置了阿里云 OSS 的连接参数。然后创建了一个 OSS Sink 对象，并指定了 OSS 存储桶的路径、存储桶名称、Writer 配置等信息。最后将数据流通过 addSink 方法写入到 OSS 中。

需要注意的是，上述示例代码中使用的 StreamingFileSink 是 Flink 提供的一个通用文件写入器，可以将数据写入到各种存储系统中，不仅仅是 OSS。如果需要将数据写入到其他存储系统，可以相应地更改代码中的 Sink 配置。

Flink提供了一个官方的OSS（Object Storage Service）Sink，可以将流式数据写入到阿里云OSS中。

1、在您的Flink项目中引入OSS Sink的依赖：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-oss</artifactId>
  <version>${flink.version}</version>
</dependency>

2、在Flink任务中使用OSS Sink：

import org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink;
import org.apache.flink.streaming.connectors.fs.bucketing.DateTimeBucketer;
import org.apache.flink.streaming.connectors.oss.OSSFileSystem;
import org.apache.flink.streaming.connectors.oss.OSSWriterConfig;
import org.apache.flink.streaming.connectors.oss.utils.OSSCredentialProvider;

...

OSSCredentialProvider credentialProvider = new DefaultOSSCredentialProvider("accessKeyId", "accessKeySecret", "securityToken");
OSSFileSystem oss = new OSSFileSystem(endpoint, credentialProvider);
BucketingSink<String> sink = new BucketingSink<>("oss://bucket/path");
sink.setBucketer(new DateTimeBucketer<>("yyyy-MM-dd/HH"));
sink.setWriter(new OSSWriter<>(oss, new OSSWriterConfig.Builder().withBufferSize(100).build()));
DataStream<String> stream = ...
stream.addSink(sink);

在上面的示例中，我们创建了一个BucketingSink，将数据写入到OSS中，其中endpoint是OSS服务的Endpoint，accessKeyId和accessKeySecret是OSS账号的AccessKeyId和AccessKeySecret，securityToken是OSS账号的STS Token（如果没有开启STS Token，可以省略该参数）。BucketingSink的构造函数中的oss://bucket/path是要将数据写入的OSS Bucket的路径，setBucketer方法是设置Bucket的分区规则，这里使用的是按照时间进行分区，setWriter方法则是设置OSS的Writer配置，这里设置了缓冲区大小为100。最后，将DataStream通过addSink方法写入到OSS中。

需要注意的是，在使用OSS Sink时，需要先将您的阿里云账号的AccessKeyId、AccessKeySecret和STS Token配置到DefaultOSSCredentialProvider中，同时确保Flink任务能够访问阿里云OSS服务。另外，如果需要将数据写入到OSS中，还需要确保Flink任务能够访问要写入的OSS Bucket。