hadoop: add FlinkFileSystemFactory (#159)

* hadoop: add FlinkFileSystemFactory

* hadoop: update

* hadoop: update

docs/hadoop_java_sdk_cn.md

@@ -43,6 +43,7 @@ $ make
 | ----   | ----                                 |
 | Spark  | `${SPARK_HOME}/jars`                 |
 | Presto | `${PRESTO_HOME}/plugin/hive-hadoop2` |
+| Flink | `${FLINK_HOME}/lib` |
 
 ## 配置参数
 
@@ -83,8 +84,6 @@ $ make
 
 ### 常用配置
 
-将以下配置参数加入到 Hadoop 配置文件 `core-site.xml` 中：
-
 ```xml
 <property>
   <name>fs.jfs.impl</name>
@@ -112,6 +111,14 @@ $ make
 </property>
 ```
 
+#### Hadoop 环境配置
+
+将配置参数加入到 Hadoop 配置文件 `core-site.xml` 中：
+
+#### Flink 配置
+
+将配置参数加入 `conf/flink-conf.yaml`。如果只是在 Flink 中使用 JuiceFS, 可以不在 Hadoop 环境配置 JuiceFS，只需要配置 Flink 客户端即可。
+
 ### 验证
 
 #### Hadoop

sdk/java/pom.xml

@@ -57,6 +57,9 @@
 			<resource>
 				<directory>libjfs/target</directory>
 			</resource>
+			<resource>
+				<directory>src/main/resources</directory>
+			</resource>
 		</resources>
 		<testResources>
 			<testResource>
@@ -124,6 +127,18 @@
 			</scope>
 			<version>1.2.1</version>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-hadoop-fs</artifactId>
+			<version>1.10.3</version>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-core</artifactId>
+			<version>1.10.3</version>
+			<scope>provided</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.hadoop</groupId>
 			<artifactId>hadoop-minicluster</artifactId>

sdk/java/src/main/java/io/juicefs/FlinkFileSystemFactory.java

+/*
+ * JuiceFS, Copyright (C) 2020 Juicedata, Inc.
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+package io.juicefs;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.runtime.fs.hdfs.HadoopFileSystem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+
+public class FlinkFileSystemFactory implements org.apache.flink.core.fs.FileSystemFactory {
+  private static final Logger LOG = LoggerFactory.getLogger(FlinkFileSystemFactory.class);
+  private org.apache.hadoop.conf.Configuration conf;
+
+  private static final String[] FLINK_CONFIG_PREFIXES = {"fs.", "juicefs."};
+  private String scheme;
+
+  @Override
+  public void configure(Configuration config) {
+    conf = new org.apache.hadoop.conf.Configuration();
+    if (config != null) {
+      for (String key : config.keySet()) {
+        for (String prefix : FLINK_CONFIG_PREFIXES) {
+          if (key.startsWith(prefix)) {
+            String value = config.getString(key, null);
+            if (value != null) {
+              if ("io.juicefs.JuiceFileSystem".equals(value.trim())) {
+                this.scheme = key.split("\\.")[1];
+              }
+              conf.set(key, value);
+            }
+          }
+        }
+      }
+    }
+  }
+
+  @Override
+  public String getScheme() {
+    if (scheme == null) {
+      return "jfs";
+    }
+    return scheme;
+  }
+
+  @Override
+  public FileSystem create(URI fsUri) throws IOException {
+    JuiceFileSystem fs = new JuiceFileSystem();
+    fs.initialize(fsUri, conf);
+    return new HadoopFileSystem(fs);
+  }
+}

sdk/java/src/main/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory

+# JuiceFS, Copyright (C) 2020 Juicedata, Inc.
+#
+# This program is free software: you can use, redistribute, and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# or later ("AGPL"), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+io.juicefs.FlinkFileSystemFactory
\ No newline at end of file