flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [4/5] flink git commit: [FLINK-3305] [core] Remove limited and inconsistent auto-magic for Joda Time
Date Mon, 01 Feb 2016 14:57:25 GMT
[FLINK-3305] [core] Remove limited and inconsistent auto-magic for Joda Time

The auto-magic for Joda Time was limited to very few classes. It was intransparent what
cases would be handled.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b6110dc3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b6110dc3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b6110dc3

Branch: refs/heads/master
Commit: b6110dc35a17340653a39209038041a5e28054b4
Parents: c4bc47a
Author: Stephan Ewen <sewen@apache.org>
Authored: Fri Jan 29 18:51:03 2016 +0100
Committer: Stephan Ewen <sewen@apache.org>
Committed: Mon Feb 1 14:46:06 2016 +0100

----------------------------------------------------------------------
 flink-java/pom.xml                              | 30 +++++++-------
 .../flink/api/java/ExecutionEnvironment.java    | 26 ++++++++----
 .../typeutils/runtime/kryo/Serializers.java     | 42 +++-----------------
 .../api/operators/TimestampedCollector.java     | 12 +++---
 .../runtime/KryoGenericTypeSerializerTest.scala | 37 +++++++----------
 .../api/scala/runtime/TupleSerializerTest.scala | 29 ++++----------
 6 files changed, 68 insertions(+), 108 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/b6110dc3/flink-java/pom.xml
----------------------------------------------------------------------
diff --git a/flink-java/pom.xml b/flink-java/pom.xml
index 8383a4a..a31e89d 100644
--- a/flink-java/pom.xml
+++ b/flink-java/pom.xml
@@ -70,22 +70,6 @@ under the License.
 		</dependency>
 
 		<dependency>
-			<groupId>de.javakaffee</groupId>
-			<artifactId>kryo-serializers</artifactId>
-			<version>0.27</version>
-		</dependency>
-
-		<dependency>
-			<groupId>joda-time</groupId>
-			<artifactId>joda-time</artifactId>
-		</dependency>
-
-		<dependency>
-			<groupId>org.joda</groupId>
-			<artifactId>joda-convert</artifactId>
-		</dependency>
-
-		<dependency>
 			<groupId>com.google.guava</groupId>
 			<artifactId>guava</artifactId>
 			<version>${guava.version}</version>
@@ -104,6 +88,20 @@ under the License.
 			<type>test-jar</type>
 			<scope>test</scope>
 		</dependency>
+
+		<dependency>
+			<groupId>joda-time</groupId>
+			<artifactId>joda-time</artifactId>
+			<version>2.5</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.joda</groupId>
+			<artifactId>joda-convert</artifactId>
+			<version>1.7</version>
+			<scope>test</scope>
+		</dependency>
 		
 	</dependencies>
 

http://git-wip-us.apache.org/repos/asf/flink/blob/b6110dc3/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index 10cb5e3..253ffa3 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -18,9 +18,10 @@
 
 package org.apache.flink.api.java;
 
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.*;
+import com.esotericsoftware.kryo.Serializer;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.InvalidProgramException;
@@ -46,7 +47,10 @@ import org.apache.flink.api.java.operators.DataSource;
 import org.apache.flink.api.java.operators.Operator;
 import org.apache.flink.api.java.operators.OperatorTranslation;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.*;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.java.typeutils.ValueTypeInfo;
 import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.Path;
@@ -54,14 +58,22 @@ import org.apache.flink.types.StringValue;
 import org.apache.flink.util.NumberSequenceIterator;
 import org.apache.flink.util.SplittableIterator;
 import org.apache.flink.util.Visitor;
+
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapreduce.Job;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.esotericsoftware.kryo.Serializer;
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
 
 /**
  * The ExecutionEnvironment is the context in which a program is executed. A

http://git-wip-us.apache.org/repos/asf/flink/blob/b6110dc3/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
index 6903d35..8bac729 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
@@ -23,13 +23,9 @@ import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
 import com.esotericsoftware.kryo.serializers.CollectionSerializer;
-
-import de.javakaffee.kryoserializers.jodatime.JodaDateTimeSerializer;
-import de.javakaffee.kryoserializers.jodatime.JodaIntervalSerializer;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.specific.SpecificRecordBase;
-
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
@@ -37,11 +33,13 @@ import org.apache.flink.api.java.Utils;
 import org.apache.flink.api.java.typeutils.GenericTypeInfo;
 import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.joda.time.DateTime;
-import org.joda.time.Interval;
 
 import java.io.Serializable;
-import java.lang.reflect.*;
+import java.lang.reflect.Field;
+import java.lang.reflect.GenericArrayType;
+import java.lang.reflect.Modifier;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -165,36 +163,6 @@ public class Serializers {
 	//	ClassTag<SpecificRecordBase> tag = scala.reflect.ClassTag$.MODULE$.apply(avroType);
 	//	reg.registerTypeWithKryoSerializer(avroType, com.twitter.chill.avro.AvroSerializer.SpecificRecordSerializer(tag));
 	}
-	
-	/**
-	 * Currently, the following classes of JodaTime are supported:
-	 *      - DateTime
-	 *      - Interval
-	 *
-	 *      The following chronologies are supported: (see {@link de.javakaffee.kryoserializers.jodatime.JodaDateTimeSerializer})
-	 * <ul>
-	 * <li>{@link org.joda.time.chrono.ISOChronology}</li>
-	 * <li>{@link org.joda.time.chrono.CopticChronology}</li>
-	 * <li>{@link org.joda.time.chrono.EthiopicChronology}</li>
-	 * <li>{@link org.joda.time.chrono.GregorianChronology}</li>
-	 * <li>{@link org.joda.time.chrono.JulianChronology}</li>
-	 * <li>{@link org.joda.time.chrono.IslamicChronology}</li>
-	 * <li>{@link org.joda.time.chrono.BuddhistChronology}</li>
-	 * <li>{@link org.joda.time.chrono.GJChronology}</li>
-	 * </ul>
-	 */
-	public static void registerJodaTime(ExecutionConfig reg) {
-		reg.registerTypeWithKryoSerializer(DateTime.class, JodaDateTimeSerializer.class);
-		reg.registerTypeWithKryoSerializer(Interval.class, JodaIntervalSerializer.class);
-	}
-	
-	/**
-	 * Register less frequently used serializers
-	 */
-	public static void registerJavaUtils(ExecutionConfig reg) {
-		// BitSet, Regex is already present through twitter-chill.
-	}
-
 
 	// --------------------------------------------------------------------------------------------
 	// Custom Serializers

http://git-wip-us.apache.org/repos/asf/flink/blob/b6110dc3/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
index 62514fc..5af5109 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
@@ -15,17 +15,18 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.flink.streaming.api.operators;
 
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.util.Collector;
-import org.joda.time.Instant;
 
 /**
  * Wrapper around an {@link Output} for user functions that expect a {@link Collector}.
  * Before giving the {@link TimestampedCollector} to a user function you must set
- * the {@link Instant timestamp} that should be attached to emitted elements. Most operators
- * would set the {@link Instant timestamp} of the incoming {@link org.apache.flink.streaming.runtime.streamrecord.StreamRecord}
here.
+ * the timestamp that should be attached to emitted elements. Most operators
+ * would set the timestamp of the incoming
+ * {@link org.apache.flink.streaming.runtime.streamrecord.StreamRecord} here.
  *
  * @param <T> The type of the elments that can be emitted.
  */
@@ -52,8 +53,9 @@ public class TimestampedCollector<T> implements Collector<T>
{
 	}
 
 	/**
-	 * Sets the {@link Instant timestamp} that is attached to elements that get emitted using
-	 * {@link #collect}
+	 * Sets the timestamp (long milliseconds) that is attached to elements that get emitted
using
+	 * {@link #collect(Object)}
+	 * 
 	 * @param timestamp The timestamp in milliseconds
 	 */
 	public void setTimestamp(long timestamp) {

http://git-wip-us.apache.org/repos/asf/flink/blob/b6110dc3/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
index 859ad2d..08a0a96 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
@@ -17,17 +17,18 @@
  */
 package org.apache.flink.api.scala.runtime
 
+import com.esotericsoftware.kryo.{Kryo, Serializer}
+import com.esotericsoftware.kryo.io.{Input, Output}
+
 import org.apache.flink.api.common.ExecutionConfig
 import org.apache.flink.api.common.typeutils.SerializerTestInstance
 import org.apache.flink.api.java.typeutils.GenericTypeInfo
-import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers
+
+import org.joda.time.LocalDate
+
 import org.junit.Test
+
 import scala.reflect._
-import org.joda.time.{DateTime, LocalDate}
-import com.esotericsoftware.kryo.Serializer
-import com.esotericsoftware.kryo.Kryo
-import com.esotericsoftware.kryo.io.Output
-import com.esotericsoftware.kryo.io.Input
 
 class KryoGenericTypeSerializerTest {
 
@@ -84,56 +85,49 @@ class KryoGenericTypeSerializerTest {
   }
 
   @Test
-  def testThrowableSerialization: Unit = {
+  def testThrowableSerialization(): Unit = {
     val a = List(new RuntimeException("Hello"), new RuntimeException("there"))
 
     runTests(a)
   }
 
   @Test
-  def jodaSerialization: Unit = {
-    val a = List(new DateTime(1), new DateTime(2))
-
-    runTests(a)
-  }
-
-  @Test
-  def jodaSerialization1: Unit = {
+  def jodaSerialization(): Unit = {
     val a = List(new LocalDate(1), new LocalDate(2))
     
     runTests(a)
   }
 
   @Test
-  def testScalaListSerialization: Unit = {
+  def testScalaListSerialization(): Unit = {
     val a = List(42,1,49,1337)
 
     runTests(a)
   }
 
   @Test
-  def testScalaMutablelistSerialization: Unit = {
+  def testScalaMutablelistSerialization(): Unit = {
     val a = scala.collection.mutable.ListBuffer(42,1,49,1337)
 
     runTests(a)
   }
 
   @Test
-  def testScalaMapSerialization: Unit = {
+  def testScalaMapSerialization(): Unit = {
     val a = Map(("1" -> 1), ("2" -> 2), ("42" -> 42), ("1337" -> 1337))
 
     runTests(Seq(a))
   }
 
   @Test
-  def testMutableMapSerialization: Unit ={
+  def testMutableMapSerialization(): Unit ={
     val a = scala.collection.mutable.Map((1 -> "1"), (2 -> "2"), (3 -> "3"))
 
     runTests(Seq(a))
   }
 
   @Test
-  def testScalaListComplexTypeSerialization: Unit = {
+  def testScalaListComplexTypeSerialization(): Unit = {
     val a = ComplexType("1234", 42, List(1,2,3,4))
     val b = ComplexType("4321", 24, List(4,3,2,1))
     val c = ComplexType("1337", 1, List(1))
@@ -143,7 +137,7 @@ class KryoGenericTypeSerializerTest {
   }
 
   @Test
-  def testHeterogenousScalaList: Unit = {
+  def testHeterogenousScalaList(): Unit = {
     val a = new DerivedType("foo", "bar")
     val b = new BaseType("foobar")
     val c = new DerivedType2("bar", "foo")
@@ -201,7 +195,6 @@ class KryoGenericTypeSerializerTest {
     // Register the custom Kryo Serializer
     val conf = new ExecutionConfig
     conf.registerTypeWithKryoSerializer(classOf[LocalDate], classOf[LocalDateSerializer])
-    Serializers.registerJodaTime(conf)
     val typeInfo = new GenericTypeInfo[T](clsTag.runtimeClass.asInstanceOf[Class[T]])
     val serializer = typeInfo.createSerializer(conf)
     val typeClass = typeInfo.getTypeClass

http://git-wip-us.apache.org/repos/asf/flink/blob/b6110dc3/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleSerializerTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleSerializerTest.scala
b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleSerializerTest.scala
index c436d62..368204b 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleSerializerTest.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleSerializerTest.scala
@@ -18,20 +18,21 @@
 package org.apache.flink.api.scala.runtime
 
 import java.util
+import java.util.Random
+
+import org.apache.flink.api.scala._
 import org.apache.flink.api.common.ExecutionConfig
-import org.apache.flink.api.java.ExecutionEnvironment
 import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
 import org.apache.flink.api.java.typeutils.runtime.AbstractGenericTypeSerializerTest._
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.typeutils.runtime.kryo.{Serializers, KryoSerializer}
 import org.apache.flink.util.StringUtils
-import org.joda.time.DateTime
+
 import org.joda.time.LocalDate
+
 import org.junit.Assert
 import org.junit.Test
-import org.apache.flink.api.scala._
+
 import scala.collection.JavaConverters._
-import java.util.Random
 
 class TupleSerializerTest {
 
@@ -98,20 +99,6 @@ class TupleSerializerTest {
     val rnd: Random = new Random(807346528946L)
 
     val testTuples = Array(
-      (StringUtils.getRandomString(rnd, 10, 100), new DateTime(rnd.nextInt)),
-      (StringUtils.getRandomString(rnd, 10, 100), new DateTime(rnd.nextInt)),
-      (StringUtils.getRandomString(rnd, 10, 100), new DateTime(rnd.nextInt)),
-      ("", rnd.nextDouble),
-      (StringUtils.getRandomString(rnd, 10, 100), new DateTime(rnd.nextInt)),
-      (StringUtils.getRandomString(rnd, 10, 100), new DateTime(rnd.nextInt)))
-    runTests(testTuples)
-  }
-
-  @Test
-  def testTuple2StringJodaTime2(): Unit = {
-    val rnd: Random = new Random(807346528946L)
-
-    val testTuples = Array(
       (StringUtils.getRandomString(rnd, 10, 100), new LocalDate(rnd.nextInt)),
       (StringUtils.getRandomString(rnd, 10, 100), new LocalDate(rnd.nextInt)),
       (StringUtils.getRandomString(rnd, 10, 100), new LocalDate(rnd.nextInt)),
@@ -208,9 +195,9 @@ class TupleSerializerTest {
   private final def runTests[T <: Product : TypeInformation](instances: Array[T]) {
     try {
       // Register the custom Kryo Serializer
-      val conf = new ExecutionConfig
+      val conf = new ExecutionConfig()
       conf.registerTypeWithKryoSerializer(classOf[LocalDate], classOf[LocalDateSerializer])
-      Serializers.registerJodaTime(conf)
+      
       val tupleTypeInfo = implicitly[TypeInformation[T]].asInstanceOf[TupleTypeInfoBase[T]]
       val serializer = tupleTypeInfo.createSerializer(conf)
       val tupleClass = tupleTypeInfo.getTypeClass


Mime
View raw message