import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Override
public TweetFileInputSplit[] createInputSplits(int minNumSplits) throws IOException {
    FileSystem fileSystem = getFileSystem();
    FileStatus[] statuses = fileSystem.listStatus(new Path(inputPath));
    logger.info("Found {} files", statuses.length);

    List<TweetFileInputSplit> splits = new ArrayList<>();
    for (int i = 0; i < statuses.length; i++) {
        FileStatus status = statuses[i];
        String fileName = status.getPath().getName();
        if (fileName.endsWith("edges")) {
            splits.add(new TweetFileInputSplit(i, status.getPath()));
        }
    }

    logger.info("Result number of splits: {}", splits.size());
    return splits.toArray(new TweetFileInputSplit[splits.size()]);
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
/**
 * Creates a new registry that writes its files to the given FileSystem at
 * the given working directory path.
 * 
 * <p>The initialization will attempt to write to the given working directory, in
 * order to catch setup/configuration errors early.
 *
 * @param fileSystem The FileSystem to use for the marker files.
 * @param workingDirectory The working directory for files to track the job status.
 *
 * @throws IOException Thrown, if the specified directory cannot be accessed.
 */
public FsNegativeRunningJobsRegistry(FileSystem fileSystem, Path workingDirectory) throws IOException {
	this.fileSystem = checkNotNull(fileSystem, "fileSystem");
	this.basePath = checkNotNull(workingDirectory, "workingDirectory");

	// to be safe, attempt to write to the working directory, to
	// catch problems early
	final Path testFile = new Path(workingDirectory, ".registry_test");
	try {
		createFile(testFile, false);
	}
	catch (IOException e) {
		throw new IOException("Unable to write to working directory: " + workingDirectory, e);
	}
	finally {
		fileSystem.delete(testFile, false);
	}
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
public CliFrontend(
		Configuration configuration,
		List<CustomCommandLine<?>> customCommandLines) throws Exception {
	this.configuration = Preconditions.checkNotNull(configuration);
	this.customCommandLines = Preconditions.checkNotNull(customCommandLines);

	try {
		FileSystem.initialize(this.configuration);
	} catch (IOException e) {
		throw new Exception("Error while setting the default " +
			"filesystem scheme from configuration.", e);
	}

	this.customCommandLineOptions = new Options();

	for (CustomCommandLine<?> customCommandLine : customCommandLines) {
		customCommandLine.addGeneralOptions(customCommandLineOptions);
		customCommandLine.addRunOptions(customCommandLineOptions);
	}

	this.clientTimeout = AkkaUtils.getClientTimeout(this.configuration);
	this.defaultParallelism = configuration.getInteger(
		ConfigConstants.DEFAULT_PARALLELISM_KEY,
		ConfigConstants.DEFAULT_PARALLELISM);
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testCompression() throws Exception {
	// given
	final Path outputPath = new Path(File.createTempFile("avro-output-file", "avro").getAbsolutePath());
	final AvroOutputFormat<User> outputFormat = new AvroOutputFormat<>(outputPath, User.class);
	outputFormat.setWriteMode(FileSystem.WriteMode.OVERWRITE);

	final Path compressedOutputPath = new Path(File.createTempFile("avro-output-file", "compressed.avro").getAbsolutePath());
	final AvroOutputFormat<User> compressedOutputFormat = new AvroOutputFormat<>(compressedOutputPath, User.class);
	compressedOutputFormat.setWriteMode(FileSystem.WriteMode.OVERWRITE);
	compressedOutputFormat.setCodec(Codec.SNAPPY);

	// when
	output(outputFormat);
	output(compressedOutputFormat);

	// then
	assertTrue(fileSize(outputPath) > fileSize(compressedOutputPath));

	// cleanup
	FileSystem fs = FileSystem.getLocalFileSystem();
	fs.delete(outputPath, false);
	fs.delete(compressedOutputPath, false);
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
/**
 *  register cache files in program level
 * @param entry contains all relevant information
 * @param name user defined name of that file
 * @throws java.io.IOException
 */
public void registerCachedFile(String name, DistributedCacheEntry entry) throws IOException {
	if (!this.cacheFile.containsKey(name)) {
		try {
			URI u = new URI(entry.filePath);
			if (!u.getPath().startsWith("/")) {
				u = new File(entry.filePath).toURI();
			}
			FileSystem fs = FileSystem.get(u);
			if (fs.exists(new Path(u.getPath()))) {
				this.cacheFile.put(name, new DistributedCacheEntry(u.toString(), entry.isExecutable));
			} else {
				throw new IOException("File " + u.toString() + " doesn't exist.");
			}
		} catch (URISyntaxException ex) {
			throw new IOException("Invalid path: " + entry.filePath, ex);
		}
	} else {
		throw new IOException("cache file " + name + "already exists!");
	}
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
	// check whether credentials exist
	Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
	Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
	Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);

	// initialize configuration with valid credentials
	final Configuration conf = new Configuration();
	conf.setString("s3.access.key", ACCESS_KEY);
	conf.setString("s3.secret.key", SECRET_KEY);
	FileSystem.initialize(conf);

	// check for uniqueness of the test directory
	final Path directory = new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
	final FileSystem fs = directory.getFileSystem();

	// directory must not yet exist
	assertFalse(fs.exists(directory));

	// reset configuration
	FileSystem.initialize(new Configuration());

	skipTest = false;
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
/**
 * Validates that even empty streams create a file and a file state handle.
 */
@Test
public void testEmptyState() throws Exception {
	final FileSystem fs = FileSystem.getLocalFileSystem();
	final Path path = new Path(new Path(tmp.newFolder().toURI()), "myFileName");

	final FileStateHandle handle;
	try (FixFileFsStateOutputStream stream = new FixFileFsStateOutputStream(fs, path)) {
		handle = stream.closeAndGetHandle();
	}

	// must have created a handle
	assertNotNull(handle);
	assertEquals(path, handle.getFilePath());

	// the pointer path should exist as a directory
	assertTrue(fs.exists(handle.getFilePath()));
	assertFalse(fs.getFileStatus(path).isDir());

	// the contents should be empty
	try (FSDataInputStream in = handle.openInputStream()) {
		assertEquals(-1, in.read());
	}
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public StateTable<K, N, SV> deserialize(
		String stateName,
		HeapKeyedStateBackend<K> stateBackend) throws IOException {

	final FileSystem fs = getFilePath().getFileSystem();
	try (FSDataInputStream inStream = fs.open(getFilePath())) {
		final DataInputViewStreamWrapper inView = new DataInputViewStreamWrapper(inStream);
		AbstractMigrationRestoreStrategy<K, N, SV> restoreStrategy =
				new AbstractMigrationRestoreStrategy<K, N, SV>(keySerializer, namespaceSerializer, stateSerializer) {
					@Override
					protected DataInputView openDataInputView() throws IOException {
						return inView;
					}
				};
		return restoreStrategy.deserialize(stateName, stateBackend);
	}
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testRenameToNonEmptyTargetDir() throws IOException {
	final FileSystem fs = FileSystem.getLocalFileSystem();

	// a source folder with a file
	final File srcFolder = temporaryFolder.newFolder();
	final File srcFile = new File(srcFolder, "someFile.txt");
	assertTrue(srcFile.createNewFile());

	// a non-empty destination folder
	final File dstFolder = temporaryFolder.newFolder();
	final File dstFile  = new File(dstFolder, "target");
	assertTrue(dstFile.createNewFile());

	// this cannot succeed because the destination folder is not empty
	assertFalse(fs.rename(new Path(srcFolder.toURI()), new Path(dstFolder.toURI())));

	// retry after deleting the occupying target file
	assertTrue(dstFile.delete());
	assertTrue(fs.rename(new Path(srcFolder.toURI()), new Path(dstFolder.toURI())));
	assertTrue(new File(dstFolder, srcFile.getName()).exists());
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Override
public void open(TweetFileInputSplit split) throws IOException {
    FileSystem fileSystem = getFileSystem();
    this.reader = new BufferedReader(new InputStreamReader(fileSystem.open(split.getPath())));
    // Pre-read next line to easily check if we've reached the end of an input split
    this.nextLine = reader.readLine();
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
/**
 * Tests loading with unexpected magic number.
 */
@Test
public void testUnexpectedSavepoint() throws Exception {
	FsSavepointStore store = new FsSavepointStore(tmp.getRoot().getPath(), "fs-savepoint-store-test-");

	// Random file
	Path filePath = new Path(tmp.getRoot().getPath(), UUID.randomUUID().toString());
	FSDataOutputStream fdos = FileSystem.get(filePath.toUri()).create(filePath, false);
	DataOutputStream dos = new DataOutputStream(fdos);
	for (int i = 0; i < 10; i++) {
		dos.writeLong(ThreadLocalRandom.current().nextLong());
	}

	try {
		store.loadSavepoint(filePath.toString());
		fail("Did not throw expected Exception");
	} catch (RuntimeException e) {
		assertTrue(e.getMessage().contains("Flink 1.0") && e.getMessage().contains("Unexpected magic number"));
	}
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testUnboundedTupleSourceAndReturnTuple() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	DataStream<Tuple4<Integer, String, Double, Long>> input = env
		.addSource(new RandomTupleSource(5).closeDelay(1500)).keyBy(1);

	DataStream<Tuple4<Long, Integer, String, Double>> output = SiddhiCEP
		.define("inputStream", input, "id", "name", "price", "timestamp")
		.cql("from inputStream select timestamp, id, name, price insert into  outputStream")
		.returns("outputStream");

	String resultPath = tempFolder.newFile().toURI().toString();
	output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
	env.execute();
	assertEquals(5, getLineCount(resultPath));
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testUnboundedPojoStreamAndReturnMap() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.setParallelism(1);
	env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
	DataStream<Event> input = env.addSource(new RandomEventSource(5));

	DataStream<Map<String, Object>> output = SiddhiCEP
		.define("inputStream", input, "id", "name", "price", "timestamp")
		.cql("from inputStream select timestamp, id, name, price insert into  outputStream")
		.returnAsMap("outputStream");

	String resultPath = tempFolder.newFile().toURI().toString();
	output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
	env.execute();
	assertEquals(5, getLineCount(resultPath));
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testUnboundedPojoStreamAndReturnPojo() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	DataStream<Event> input = env.addSource(new RandomEventSource(5));
	input.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Event>() {
		@Override
		public long extractAscendingTimestamp(Event element) {
			return element.getTimestamp();
		}
	});

	DataStream<Event> output = SiddhiCEP
		.define("inputStream", input, "id", "name", "price", "timestamp")
		.cql("from inputStream select timestamp, id, name, price insert into  outputStream")
		.returns("outputStream", Event.class);

	String resultPath = tempFolder.newFile().toURI().toString();
	output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
	env.execute();
	assertEquals(5, getLineCount(resultPath));
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testMultipleUnboundedPojoStreamSimpleUnion() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	DataStream<Event> input1 = env.addSource(new RandomEventSource(2), "input1");
	DataStream<Event> input2 = env.addSource(new RandomEventSource(2), "input2");
	DataStream<Event> input3 = env.addSource(new RandomEventSource(2), "input2");
	DataStream<Event> output = SiddhiCEP
		.define("inputStream1", input1, "id", "name", "price", "timestamp")
		.union("inputStream2", input2, "id", "name", "price", "timestamp")
		.union("inputStream3", input3, "id", "name", "price", "timestamp")
		.cql(
			"from inputStream1 select timestamp, id, name, price insert into outputStream;"
				+ "from inputStream2 select timestamp, id, name, price insert into outputStream;"
				+ "from inputStream3 select timestamp, id, name, price insert into outputStream;"
		)
		.returns("outputStream", Event.class);

	String resultPath = tempFolder.newFile().toURI().toString();
	output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
	env.execute();
	assertEquals(6, getLineCount(resultPath));
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testEmptyState() throws Exception {
	FsCheckpointStreamFactory.CheckpointStateOutputStream stream =
			new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(TEMP_DIR_PATH, FileSystem.getLocalFileSystem(), 1024, 512);

	StreamStateHandle handle = stream.closeAndGetHandle();
	assertTrue(handle == null);
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testUnboundedPojoSourceAndReturnTuple() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    DataStream<Event> input = env.addSource(new RandomEventSource(5));

    DataStream<Tuple4<Long, Integer, String, Double>> output = SiddhiCEP
        .define("inputStream", input, "id", "name", "price", "timestamp")
        .cql("from inputStream select timestamp, id, name, price insert into  outputStream")
        .returns("outputStream");

    DataStream<Integer> following = output.map(new MapFunction<Tuple4<Long, Integer, String, Double>, Integer>() {
        @Override
        public Integer map(Tuple4<Long, Integer, String, Double> value) throws Exception {
            return value.f1;
        }
    });
    String resultPath = tempFolder.newFile().toURI().toString();
    following.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
    env.execute();
    assertEquals(5, getLineCount(resultPath));
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testUnboundedTupleSourceAndReturnTuple() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    DataStream<Tuple4<Integer, String, Double, Long>> input = env
        .addSource(new RandomTupleSource(5).closeDelay(1500)).keyBy(1);

    DataStream<Tuple4<Long, Integer, String, Double>> output = SiddhiCEP
        .define("inputStream", input, "id", "name", "price", "timestamp")
        .cql("from inputStream select timestamp, id, name, price insert into  outputStream")
        .returns("outputStream");

    String resultPath = tempFolder.newFile().toURI().toString();
    output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
    env.execute();
    assertEquals(5, getLineCount(resultPath));
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testSimpleFileWriteAndRead() throws Exception {
	final Configuration conf = new Configuration();
	conf.setString("s3.access.key", ACCESS_KEY);
	conf.setString("s3.secret.key", SECRET_KEY);

	final String testLine = "Hello Upload!";

	FileSystem.initialize(conf);

	final Path path = new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR + "/test.txt");
	final FileSystem fs = path.getFileSystem();

	try {
		try (FSDataOutputStream out = fs.create(path, WriteMode.OVERWRITE);
				OutputStreamWriter writer = new OutputStreamWriter(out, StandardCharsets.UTF_8)) {
			writer.write(testLine);
		}

		try (FSDataInputStream in = fs.open(path);
				InputStreamReader ir = new InputStreamReader(in, StandardCharsets.UTF_8);
				BufferedReader reader = new BufferedReader(ir)) {
			String line = reader.readLine();
			assertEquals(testLine, line);
		}
	}
	finally {
		fs.delete(path, false);
	}
}
 

import org.apache.flink.core.fs.FileSystem; //导入依赖的package包/类
@Test
public void testConfigPropagation() throws Exception{
	final Configuration conf = new Configuration();
	conf.setString("s3.access-key", "test_access_key_id");
	conf.setString("s3.secret-key", "test_secret_access_key");

	FileSystem.initialize(conf);

	FileSystem fs = FileSystem.get(new URI("s3://test"));
	validateBasicCredentials(fs);
}