@Override
	public JavaDStream<String> process(JavaDStream<String> input) {
		input.foreachRDD(new Function<JavaRDD<String>, Void>() {

			@Override
			public Void call(JavaRDD<String> rdd) {
				rdd.foreachPartition(new VoidFunction<Iterator<String>>() {

					@Override
					public void call(Iterator<String> items) throws Exception {
						FileWriter fw;
						BufferedWriter bw = null;
						try {
							fw = new FileWriter(file.getAbsoluteFile());
							bw = new BufferedWriter(fw);
							while (items.hasNext()) {
								bw.append(items.next() + System.lineSeparator());
							}
						}
						catch (IOException ioe) {
							throw new RuntimeException(ioe);
						}
						finally {
							if (bw != null) {
								bw.close();
							}
						}
					}
				});
				return null;
			}
		});
		return null;
	}
@author Mark Fisher
@author Ilayaperumal Gopinathan
@since 1.1
spring-xd-spark-streaming/src/main/java/org/springframework/xd/spark/streaming/examples/java/FileLogger.java