You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2015/02/12 14:21:35 UTC
flink git commit: [FLINK-1510] Make AvroInputFormat splittable
Repository: flink
Updated Branches:
refs/heads/master 46a675ccb -> 3241214c7
[FLINK-1510] Make AvroInputFormat splittable
This closes #387
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3241214c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3241214c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3241214c
Branch: refs/heads/master
Commit: 3241214c7eb09f5f091e894eb04462ed41111672
Parents: 46a675c
Author: Robert Metzger <rm...@apache.org>
Authored: Wed Feb 11 09:51:12 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Thu Feb 12 14:20:57 2015 +0100
----------------------------------------------------------------------
.../flink/api/common/io/FileInputFormat.java | 2 +-
.../runtime/fs/hdfs/HadoopDataInputStream.java | 5 +-
.../flink/runtime/fs/hdfs/HadoopFileSystem.java | 9 +-
.../flink/runtime/fs/maprfs/MapRFileSystem.java | 11 +-
.../flink/runtime/fs/s3/S3DataInputStream.java | 1 +
flink-staging/flink-avro/pom.xml | 2 +-
.../api/avro/FSDataInputStreamWrapper.java | 8 +-
.../flink/api/java/io/AvroInputFormat.java | 23 ++-
.../io/avro/AvroSplittableInputFormatTest.java | 202 +++++++++++++++++++
9 files changed, 233 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
index ed92eed..82e4c54 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
@@ -566,7 +566,7 @@ public abstract class FileInputFormat<OT> implements InputFormat<OT, FileInputSp
return length;
}
- private boolean testForUnsplittable(FileStatus pathFile) {
+ protected boolean testForUnsplittable(FileStatus pathFile) {
if(pathFile.getPath().getName().endsWith(DEFLATE_SUFFIX)) {
unsplittable = true;
return true;
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataInputStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataInputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataInputStream.java
index 8aa3c57..9a62885 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataInputStream.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataInputStream.java
@@ -42,6 +42,7 @@ public final class HadoopDataInputStream extends FSDataInputStream {
this.fsDataInputStream = fsDataInputStream;
}
+
@Override
public synchronized void seek(long desired) throws IOException {
@@ -50,25 +51,21 @@ public final class HadoopDataInputStream extends FSDataInputStream {
@Override
public long getPos() throws IOException {
-
return fsDataInputStream.getPos();
}
@Override
public int read() throws IOException {
-
return fsDataInputStream.read();
}
@Override
public void close() throws IOException {
-
fsDataInputStream.close();
}
@Override
public int read(byte[] buffer, int offset, int length) throws IOException {
-
return fsDataInputStream.read(buffer, offset, length);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
index 37061bf..1a79f23 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
@@ -364,16 +364,15 @@ public final class HadoopFileSystem extends FileSystem implements HadoopFileSyst
@Override
public FSDataInputStream open(final Path f, final int bufferSize) throws IOException {
-
- final org.apache.hadoop.fs.FSDataInputStream fdis = this.fs.open(new org.apache.hadoop.fs.Path(f.toString()),
- bufferSize);
-
+ final org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(f.toString());
+ final org.apache.hadoop.fs.FSDataInputStream fdis = this.fs.open(path, bufferSize);
return new HadoopDataInputStream(fdis);
}
@Override
public FSDataInputStream open(final Path f) throws IOException {
- final org.apache.hadoop.fs.FSDataInputStream fdis = fs.open(new org.apache.hadoop.fs.Path(f.toString()));
+ final org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(f.toString());
+ final org.apache.hadoop.fs.FSDataInputStream fdis = fs.open(path);
return new HadoopDataInputStream(fdis);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
index 11216da..41bbaca 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/maprfs/MapRFileSystem.java
@@ -297,18 +297,15 @@ public final class MapRFileSystem extends FileSystem {
@Override
public FSDataInputStream open(final Path f, final int bufferSize)
throws IOException {
-
- final org.apache.hadoop.fs.FSDataInputStream fdis = this.fs.open(
- new org.apache.hadoop.fs.Path(f.toString()), bufferSize);
-
+ final org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(f.toString());
+ final org.apache.hadoop.fs.FSDataInputStream fdis = this.fs.open(path, bufferSize);
return new HadoopDataInputStream(fdis);
}
@Override
public FSDataInputStream open(final Path f) throws IOException {
-
- final org.apache.hadoop.fs.FSDataInputStream fdis = this.fs
- .open(new org.apache.hadoop.fs.Path(f.toString()));
+ final org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(f.toString());
+ final org.apache.hadoop.fs.FSDataInputStream fdis = this.fs.open(path);
return new HadoopDataInputStream(fdis);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-runtime/src/main/java/org/apache/flink/runtime/fs/s3/S3DataInputStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/s3/S3DataInputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/s3/S3DataInputStream.java
index 715f1a1..c839640 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/s3/S3DataInputStream.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/s3/S3DataInputStream.java
@@ -50,6 +50,7 @@ public class S3DataInputStream extends FSDataInputStream {
*/
private long marked;
+
/**
* Constructs a new input stream which reads its data from the specified S3 object.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-staging/flink-avro/pom.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-avro/pom.xml b/flink-staging/flink-avro/pom.xml
index 15bb801..90a7be2 100644
--- a/flink-staging/flink-avro/pom.xml
+++ b/flink-staging/flink-avro/pom.xml
@@ -47,7 +47,7 @@ under the License.
<artifactId>flink-clients</artifactId>
<version>${project.version}</version>
</dependency>
-
+
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-staging/flink-avro/src/main/java/org/apache/flink/api/avro/FSDataInputStreamWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-avro/src/main/java/org/apache/flink/api/avro/FSDataInputStreamWrapper.java b/flink-staging/flink-avro/src/main/java/org/apache/flink/api/avro/FSDataInputStreamWrapper.java
index 655b6ff..709c4f1 100644
--- a/flink-staging/flink-avro/src/main/java/org/apache/flink/api/avro/FSDataInputStreamWrapper.java
+++ b/flink-staging/flink-avro/src/main/java/org/apache/flink/api/avro/FSDataInputStreamWrapper.java
@@ -33,17 +33,17 @@ import org.apache.flink.core.fs.FSDataInputStream;
*/
public class FSDataInputStreamWrapper implements Closeable, SeekableInput {
private final FSDataInputStream stream;
- private final long len;
private long pos;
+ private long len;
public FSDataInputStreamWrapper(FSDataInputStream stream, long len) {
this.stream = stream;
- this.len = len;
this.pos = 0;
+ this.len = len;
}
- public long length() {
- return len;
+ public long length() throws IOException {
+ return this.len;
}
public int read(byte[] b, int off, int len) throws IOException {
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-staging/flink-avro/src/main/java/org/apache/flink/api/java/io/AvroInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-avro/src/main/java/org/apache/flink/api/java/io/AvroInputFormat.java b/flink-staging/flink-avro/src/main/java/org/apache/flink/api/java/io/AvroInputFormat.java
index fbb0f4c..09fcacb 100644
--- a/flink-staging/flink-avro/src/main/java/org/apache/flink/api/java/io/AvroInputFormat.java
+++ b/flink-staging/flink-avro/src/main/java/org/apache/flink/api/java/io/AvroInputFormat.java
@@ -44,20 +44,19 @@ public class AvroInputFormat<E> extends FileInputFormat<E> implements ResultType
private static final long serialVersionUID = 1L;
private static final Logger LOG = LoggerFactory.getLogger(AvroInputFormat.class);
-
-
+
private final Class<E> avroValueType;
private boolean reuseAvroValue = true;
-
private transient FileReader<E> dataFileReader;
+ private transient long end;
+
public AvroInputFormat(Path filePath, Class<E> type) {
super(filePath);
this.avroValueType = type;
- this.unsplittable = true;
}
@@ -70,6 +69,13 @@ public class AvroInputFormat<E> extends FileInputFormat<E> implements ResultType
public void setReuseAvroValue(boolean reuseAvroValue) {
this.reuseAvroValue = reuseAvroValue;
}
+
+ /**
+ * If set, the InputFormat will only read entire files.
+ */
+ public void setUnsplittable(boolean unsplittable) {
+ this.unsplittable = unsplittable;
+ }
// --------------------------------------------------------------------------------------------
// Typing
@@ -97,20 +103,21 @@ public class AvroInputFormat<E> extends FileInputFormat<E> implements ResultType
LOG.info("Opening split " + split);
- SeekableInput in = new FSDataInputStreamWrapper(stream, (int) split.getLength());
-
+ SeekableInput in = new FSDataInputStreamWrapper(stream, split.getPath().getFileSystem().getFileStatus(split.getPath()).getLen());
+
dataFileReader = DataFileReader.openReader(in, datumReader);
dataFileReader.sync(split.getStart());
+ this.end = split.getStart() + split.getLength();
}
@Override
public boolean reachedEnd() throws IOException {
- return !dataFileReader.hasNext();
+ return !dataFileReader.hasNext() || dataFileReader.pastSync(end);
}
@Override
public E nextRecord(E reuseValue) throws IOException {
- if (!dataFileReader.hasNext()) {
+ if (reachedEnd()) {
return null;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3241214c/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroSplittableInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroSplittableInputFormatTest.java b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroSplittableInputFormatTest.java
new file mode 100644
index 0000000..0d60eee
--- /dev/null
+++ b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroSplittableInputFormatTest.java
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.io.avro;
+
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.io.DatumWriter;
+import org.apache.avro.specific.SpecificDatumWriter;
+import org.apache.flink.api.io.avro.generated.Colors;
+import org.apache.flink.api.io.avro.generated.User;
+import org.apache.flink.api.java.io.AvroInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.core.fs.Path;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test the avro input format.
+ * (The testcase is mostly the getting started tutorial of avro)
+ * http://avro.apache.org/docs/current/gettingstartedjava.html
+ */
+public class AvroSplittableInputFormatTest {
+
+ private File testFile;
+
+ final static String TEST_NAME = "Alyssa";
+
+ final static String TEST_ARRAY_STRING_1 = "ELEMENT 1";
+ final static String TEST_ARRAY_STRING_2 = "ELEMENT 2";
+
+ final static boolean TEST_ARRAY_BOOLEAN_1 = true;
+ final static boolean TEST_ARRAY_BOOLEAN_2 = false;
+
+ final static Colors TEST_ENUM_COLOR = Colors.GREEN;
+
+ final static String TEST_MAP_KEY1 = "KEY 1";
+ final static long TEST_MAP_VALUE1 = 8546456L;
+ final static String TEST_MAP_KEY2 = "KEY 2";
+ final static long TEST_MAP_VALUE2 = 17554L;
+
+ final static int NUM_RECORDS = 5000;
+
+ @Before
+ public void createFiles() throws IOException {
+ testFile = File.createTempFile("AvroSplittableInputFormatTest", null);
+
+ ArrayList<CharSequence> stringArray = new ArrayList<CharSequence>();
+ stringArray.add(TEST_ARRAY_STRING_1);
+ stringArray.add(TEST_ARRAY_STRING_2);
+
+ ArrayList<Boolean> booleanArray = new ArrayList<Boolean>();
+ booleanArray.add(TEST_ARRAY_BOOLEAN_1);
+ booleanArray.add(TEST_ARRAY_BOOLEAN_2);
+
+ HashMap<CharSequence, Long> longMap = new HashMap<CharSequence, Long>();
+ longMap.put(TEST_MAP_KEY1, TEST_MAP_VALUE1);
+ longMap.put(TEST_MAP_KEY2, TEST_MAP_VALUE2);
+
+
+ User user1 = new User();
+ user1.setName(TEST_NAME);
+ user1.setFavoriteNumber(256);
+ user1.setTypeDoubleTest(123.45d);
+ user1.setTypeBoolTest(true);
+ user1.setTypeArrayString(stringArray);
+ user1.setTypeArrayBoolean(booleanArray);
+ user1.setTypeEnum(TEST_ENUM_COLOR);
+ user1.setTypeMap(longMap);
+
+ // Construct via builder
+ User user2 = User.newBuilder()
+ .setName(TEST_NAME)
+ .setFavoriteColor("blue")
+ .setFavoriteNumber(null)
+ .setTypeBoolTest(false)
+ .setTypeDoubleTest(1.337d)
+ .setTypeNullTest(null)
+ .setTypeLongTest(1337L)
+ .setTypeArrayString(new ArrayList<CharSequence>())
+ .setTypeArrayBoolean(new ArrayList<Boolean>())
+ .setTypeNullableArray(null)
+ .setTypeEnum(Colors.RED)
+ .setTypeMap(new HashMap<CharSequence, Long>())
+ .build();
+ DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);
+ DataFileWriter<User> dataFileWriter = new DataFileWriter<User>(userDatumWriter);
+ dataFileWriter.create(user1.getSchema(), testFile);
+ dataFileWriter.append(user1);
+ dataFileWriter.append(user2);
+
+ Random rnd = new Random(1337);
+ for(int i = 0; i < NUM_RECORDS -2 ; i++) {
+ User user = new User();
+ user.setName(TEST_NAME + rnd.nextInt());
+ user.setFavoriteNumber(rnd.nextInt());
+ user.setTypeDoubleTest(rnd.nextDouble());
+ user.setTypeBoolTest(true);
+ user.setTypeArrayString(stringArray);
+ user.setTypeArrayBoolean(booleanArray);
+ user.setTypeEnum(TEST_ENUM_COLOR);
+ user.setTypeMap(longMap);
+
+ dataFileWriter.append(user);
+ }
+ dataFileWriter.close();
+ }
+
+ @Test
+ public void testSplittedIF() throws IOException {
+ Configuration parameters = new Configuration();
+
+ AvroInputFormat<User> format = new AvroInputFormat<User>(new Path(testFile.getAbsolutePath()), User.class);
+
+ format.configure(parameters);
+ FileInputSplit[] splits = format.createInputSplits(4);
+ assertEquals(splits.length, 4);
+ int elements = 0;
+ int elementsPerSplit[] = new int[4];
+ for(int i = 0; i < splits.length; i++) {
+ format.open(splits[i]);
+ while(!format.reachedEnd()) {
+ User u = format.nextRecord(null);
+ Assert.assertTrue(u.getName().toString().startsWith(TEST_NAME));
+ elements++;
+ elementsPerSplit[i]++;
+ }
+ format.close();
+ }
+ Assert.assertEquals(1491, elementsPerSplit[0]);
+ Assert.assertEquals(1491, elementsPerSplit[1]);
+ Assert.assertEquals(1490, elementsPerSplit[2]);
+ Assert.assertEquals(528, elementsPerSplit[3]);
+ Assert.assertEquals(NUM_RECORDS, elements);
+ format.close();
+ }
+
+ /*
+ This test is gave the reference values for the test of Flink's IF.
+
+ This dependency needs to be added
+
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro-mapred</artifactId>
+ <version>1.7.6</version>
+ </dependency>
+
+ @Test
+ public void testHadoop() throws Exception {
+ JobConf jf = new JobConf();
+ FileInputFormat.addInputPath(jf, new org.apache.hadoop.fs.Path(testFile.toURI()));
+ jf.setBoolean(org.apache.avro.mapred.AvroInputFormat.IGNORE_FILES_WITHOUT_EXTENSION_KEY, false);
+ org.apache.avro.mapred.AvroInputFormat<User> format = new org.apache.avro.mapred.AvroInputFormat<User>();
+ InputSplit[] sp = format.getSplits(jf, 4);
+ int elementsPerSplit[] = new int[4];
+ int cnt = 0;
+ int i = 0;
+ for(InputSplit s:sp) {
+ RecordReader<AvroWrapper<User>, NullWritable> r = format.getRecordReader(s, jf, new HadoopDummyReporter());
+ AvroWrapper<User> k = r.createKey();
+ NullWritable v = r.createValue();
+
+ while(r.next(k,v)) {
+ cnt++;
+ elementsPerSplit[i]++;
+ }
+ i++;
+ }
+ System.out.println("Status "+Arrays.toString(elementsPerSplit));
+ } **/
+
+ @After
+ public void deleteFiles() {
+ testFile.delete();
+ }
+}