使用Flume写入JindoFS

使用Flume写入JindoFS Block模式

# 配置JFS Sink

xxx.sinks.jfs_sink.hdfs.path = jfs://${your_ns_name}/flume_dir/%H%M/%S
xxx.sinks.jfs_sink.hdfs.round =true
xxx.sinks.jfs_sink.hdfs.roundValue = 15
xxx.sinks.jfs_sink.hdfs.Unit = minute
xxx.sinks.jfs_sink.hdfs.filePrefix = your_topic

# Sink参数，batchSize需要设置大一些，建议每次Flush的量在32MB以上，否则会影响性能。
xxx.sinks.jfs_sink.hdfs.batchSize = 100000
...
xxx.sinks.jfs_sink.rollSize = 3600
xxx.sinks.jfs_sink.threadsPoolSize = 30
xxx.sinks.jfs_sink.fileType = DataStream
xxx.sinks.jfs_sink.writeFormat = Text

使用Flume写入JindoFS Cache模式或OSS

1. 开启Flush和Recover功能。
   具体步骤，请参见开启Flush和Recover功能。
2. 配置Sink。
   各参数含义请参见Apache Flume。配置Sink示例如下：

   # 配置OSS Sink
   
   xxx.sinks.oss_sink.hdfs.path = oss://${your_bucket}/flume_dir/%H%M/%S
   xxx.sinks.oss_sink.hdfs.round = true
   xxx.sinks.oss_sink.hdfs.roundValue = 15
   xxx.sinks.oss_sink.hdfs.Unit = minute
   xxx.sinks.oss_sink.hdfs.filePrefix = <your_topic>
   
   # Sink参数，batchSize需要设置大一些，建议每次Flush的量在32MB以上，否则会影响性能。
   xxx.sinks.oss_sink.hdfs.batchSize = 100000
   ...
   xxx.sinks.oss_sink.rollSize = 3600
   xxx.sinks.oss_sink.threadsPoolSize = 30

   说明 本文示例中的your_bucket为您OSS Bucket的名称。
3. 恢复文件Flush的数据。
   具体步骤，请参见恢复文件Flush的数据。

开启Flush和Recover功能

使用Flume写入JindoFS Cache模式时，需要开启Flush和Recover功能。

1. 进入SmartData服务。
   1. 登录阿里云E-MapReduce控制台。
      
   2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
      
   3. 单击上方的集群管理页签。
      
   4. 在集群管理页面，单击相应集群所在行的详情。
      
   5. 在左侧导航栏，选择集群服务。
2. 进入smartdata-site页面。
   1. 单击配置页签。
   2. 在服务配置页面，单击storage。
3. 单击右上角的自定义配置，添加如下参数。

   参数	描述
   fs.jfs.cache.oss.flush.enable	是否开启Flush和Recover功能，开启时需要设置为true。
   fs.jfs.cache.flush.staging.path	Flush的数据和Manifest的暂存区，默认值为/tmp。 例如：在使用默认值时，如果文件的写入路径是：oss://test-bucket/dir1/file1，则Staging的路径为oss://test-bucket/tmp/dir1/file1。

4. 保存配置。
   1. 单击右上角的保存。
   2. 在确认修改对话框中，输入执行原因，开启自动更新配置。
   3. 单击确定。

恢复文件Flush的数据

jindo jfs -recover [-R]
         [-flushStagingPath {flushStagingPath}]
         [-accessKeyId ${accessKeyId}]
         [-accessKeySecret ${accessKeySecret}]
         <path>

JindoOssFileSystem jindoFileSystem = (JindoOssFileSystem) fs;
boolean isFolder = true;
jindoFileSystem.recover(path, isFolder);