You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2018/08/29 12:39:21 UTC
[flink] branch master updated: [FLINK-9962] allow users to specify
TimeZone in DateTimeBucketer
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7eb41f3 [FLINK-9962] allow users to specify TimeZone in DateTimeBucketer
7eb41f3 is described below
commit 7eb41f33fbce8da52cbba160f0ecff6fe96f124b
Author: Bowen Li <bo...@gmail.com>
AuthorDate: Sun Aug 5 01:50:03 2018 +0200
[FLINK-9962] allow users to specify TimeZone in DateTimeBucketer
This closes #6492.
---
docs/dev/connectors/filesystem_sink.md | 7 ++-
.../connectors/fs/bucketing/DateTimeBucketer.java | 54 ++++++++++++-----
.../fs/bucketing/DateTimeBucketerTest.java | 60 +++++++++++++++++++
.../bucketassigners/DateTimeBucketAssigner.java | 54 +++++++++++++----
.../DateTimeBucketAssignerTest.java | 70 ++++++++++++++++++++++
5 files changed, 215 insertions(+), 30 deletions(-)
diff --git a/docs/dev/connectors/filesystem_sink.md b/docs/dev/connectors/filesystem_sink.md
index af1349d..79ed08e 100644
--- a/docs/dev/connectors/filesystem_sink.md
+++ b/docs/dev/connectors/filesystem_sink.md
@@ -70,7 +70,8 @@ stored. The sink can be further configured by specifying a custom bucketer, writ
By default the bucketing sink will split by the current system time when elements arrive and will
use the datetime pattern `"yyyy-MM-dd--HH"` to name the buckets. This pattern is passed to
-`SimpleDateFormat` with the current system time to form a bucket path. A new bucket will be created
+`DateTimeFormatter` with the current system time and JVM's default timezone to form a bucket path.
+Users can also specify a timezone for the bucketer to format bucket path. A new bucket will be created
whenever a new date is encountered. For example, if you have a pattern that contains minutes as the
finest granularity you will get a new bucket every minute. Each bucket is itself a directory that
contains several part files: each parallel instance of the sink will create its own part file and
@@ -105,7 +106,7 @@ Example:
DataStream<Tuple2<IntWritable,Text>> input = ...;
BucketingSink<String> sink = new BucketingSink<String>("/base/path");
-sink.setBucketer(new DateTimeBucketer<String>("yyyy-MM-dd--HHmm"));
+sink.setBucketer(new DateTimeBucketer<String>("yyyy-MM-dd--HHmm", ZoneId.of("America/Los_Angeles")));
sink.setWriter(new SequenceFileWriter<IntWritable, Text>());
sink.setBatchSize(1024 * 1024 * 400); // this is 400 MB,
sink.setBatchRolloverInterval(20 * 60 * 1000); // this is 20 mins
@@ -119,7 +120,7 @@ input.addSink(sink);
val input: DataStream[Tuple2[IntWritable, Text]] = ...
val sink = new BucketingSink[String]("/base/path")
-sink.setBucketer(new DateTimeBucketer[String]("yyyy-MM-dd--HHmm"))
+sink.setBucketer(new DateTimeBucketer[String]("yyyy-MM-dd--HHmm", ZoneId.of("America/Los_Angeles")))
sink.setWriter(new SequenceFileWriter[IntWritable, Text]())
sink.setBatchSize(1024 * 1024 * 400) // this is 400 MB,
sink.setBatchRolloverInterval(20 * 60 * 1000); // this is 20 mins
diff --git a/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/DateTimeBucketer.java b/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/DateTimeBucketer.java
index b7035fe..edabb6b 100644
--- a/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/DateTimeBucketer.java
+++ b/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/DateTimeBucketer.java
@@ -19,13 +19,15 @@
package org.apache.flink.streaming.connectors.fs.bucketing;
import org.apache.flink.streaming.connectors.fs.Clock;
+import org.apache.flink.util.Preconditions;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
import java.io.ObjectInputStream;
-import java.text.SimpleDateFormat;
-import java.util.Date;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
/**
* A {@link Bucketer} that assigns to buckets based on current system time.
@@ -38,8 +40,8 @@ import java.util.Date;
* is determined based on the current system time and the user provided format string.
*
*
- * <p>{@link SimpleDateFormat} is used to derive a date string from the current system time and
- * the date format string. The default format string is {@code "yyyy-MM-dd--HH"} so the rolling
+ * <p>{@link DateTimeFormatter} is used to derive a date string from the current system time and
+ * the date format string with a timezone. The default format string is {@code "yyyy-MM-dd--HH"} so the rolling
* files will have a granularity of hours.
*
*
@@ -62,43 +64,67 @@ public class DateTimeBucketer<T> implements Bucketer<T> {
private final String formatString;
- private transient SimpleDateFormat dateFormatter;
+ private final ZoneId zoneId;
+
+ private transient DateTimeFormatter dateTimeFormatter;
/**
- * Creates a new {@code DateTimeBucketer} with format string {@code "yyyy-MM-dd--HH"}.
+ * Creates a new {@code DateTimeBucketer} with format string {@code "yyyy-MM-dd--HH"} using JVM's default timezone.
*/
public DateTimeBucketer() {
this(DEFAULT_FORMAT_STRING);
}
/**
- * Creates a new {@code DateTimeBucketer} with the given date/time format string.
+ * Creates a new {@code DateTimeBucketer} with the given date/time format string using JVM's default timezone.
*
- * @param formatString The format string that will be given to {@code SimpleDateFormat} to determine
+ * @param formatString The format string that will be given to {@code DateTimeFormatter} to determine
* the bucket path.
*/
public DateTimeBucketer(String formatString) {
- this.formatString = formatString;
+ this(formatString, ZoneId.systemDefault());
+ }
+
+ /**
+ * Creates a new {@code DateTimeBucketer} with format string {@code "yyyy-MM-dd--HH"} using the given timezone.
+ *
+ * @param zoneId The timezone used to format {@code DateTimeFormatter} for bucket path.
+ */
+ public DateTimeBucketer(ZoneId zoneId) {
+ this(DEFAULT_FORMAT_STRING, zoneId);
+ }
+
+ /**
+ * Creates a new {@code DateTimeBucketer} with the given date/time format string using the given timezone.
+ *
+ * @param formatString The format string that will be given to {@code DateTimeFormatter} to determine
+ * the bucket path.
+ * @param zoneId The timezone used to format {@code DateTimeFormatter} for bucket path.
+ */
+ public DateTimeBucketer(String formatString, ZoneId zoneId) {
+ this.formatString = Preconditions.checkNotNull(formatString);
+ this.zoneId = Preconditions.checkNotNull(zoneId);
- this.dateFormatter = new SimpleDateFormat(formatString);
+ this.dateTimeFormatter = DateTimeFormatter.ofPattern(this.formatString).withZone(zoneId);
}
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
- this.dateFormatter = new SimpleDateFormat(formatString);
+ this.dateTimeFormatter = DateTimeFormatter.ofPattern(formatString).withZone(zoneId);
}
@Override
public Path getBucketPath(Clock clock, Path basePath, T element) {
- String newDateTimeString = dateFormatter.format(new Date(clock.currentTimeMillis()));
+ String newDateTimeString = dateTimeFormatter.format(Instant.ofEpochMilli(clock.currentTimeMillis()));
return new Path(basePath + "/" + newDateTimeString);
}
@Override
public String toString() {
return "DateTimeBucketer{" +
- "formatString='" + formatString + '\'' +
- '}';
+ "formatString='" + formatString + '\'' +
+ ", zoneId=" + zoneId +
+ '}';
}
}
diff --git a/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/DateTimeBucketerTest.java b/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/DateTimeBucketerTest.java
new file mode 100644
index 0000000..3ce9f49
--- /dev/null
+++ b/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/DateTimeBucketerTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.streaming.connectors.fs.bucketing;
+
+import org.apache.flink.streaming.connectors.fs.Clock;
+
+import org.apache.hadoop.fs.Path;
+import org.junit.Test;
+
+import java.time.ZoneId;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link DateTimeBucketer}.
+ */
+public class DateTimeBucketerTest {
+ private static final long TEST_TIME_IN_MILLIS = 1533363082011L;
+ private static final Path TEST_PATH = new Path("test");
+
+ private static final Clock mockedClock = new MockedClock();
+
+ @Test
+ public void testGetBucketPathWithSpecifiedTimezone() {
+ DateTimeBucketer bucketer = new DateTimeBucketer(ZoneId.of("America/Los_Angeles"));
+
+ assertEquals(new Path("test/2018-08-03--23"), bucketer.getBucketPath(mockedClock, TEST_PATH, null));
+ }
+
+ @Test
+ public void testGetBucketPathWithSpecifiedFormatString() {
+ DateTimeBucketer bucketer = new DateTimeBucketer("yyyy-MM-dd-HH", ZoneId.of("America/Los_Angeles"));
+
+ assertEquals(new Path("test/2018-08-03-23"), bucketer.getBucketPath(mockedClock, TEST_PATH, null));
+ }
+
+ private static class MockedClock implements Clock {
+
+ @Override
+ public long currentTimeMillis() {
+ return TEST_TIME_IN_MILLIS;
+ }
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssigner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssigner.java
index 5d30f39..cb43822 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssigner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssigner.java
@@ -21,9 +21,11 @@ package org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
+import org.apache.flink.util.Preconditions;
-import java.text.SimpleDateFormat;
-import java.util.Date;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
/**
* A {@link BucketAssigner} that assigns to buckets based on current system time.
@@ -37,14 +39,14 @@ import java.util.Date;
* user provided format string.
*
*
- * <p>{@link SimpleDateFormat} is used to derive a date string from the current system time and
+ * <p>{@link DateTimeFormatter} is used to derive a date string from the current system time and
* the date format string. The default format string is {@code "yyyy-MM-dd--HH"} so the rolling
* files will have a granularity of hours.
*
* <p>Example:
*
* <pre>{@code
- * Bucketer buck = new DateTimeBucketer("yyyy-MM-dd--HH");
+ * BucketAssigner bucketAssigner = new DateTimeBucketAssigner("yyyy-MM-dd--HH");
* }</pre>
*
* <p>This will create for example the following bucket path:
@@ -60,31 +62,54 @@ public class DateTimeBucketAssigner<IN> implements BucketAssigner<IN, String> {
private final String formatString;
- private transient SimpleDateFormat dateFormatter;
+ private final ZoneId zoneId;
+
+ private transient DateTimeFormatter dateTimeFormatter;
/**
- * Creates a new {@code DateTimeBucketer} with format string {@code "yyyy-MM-dd--HH"}.
+ * Creates a new {@code DateTimeBucketAssigner} with format string {@code "yyyy-MM-dd--HH"}.
*/
public DateTimeBucketAssigner() {
this(DEFAULT_FORMAT_STRING);
}
/**
- * Creates a new {@code DateTimeBucketer} with the given date/time format string.
+ * Creates a new {@code DateTimeBucketAssigner} with the given date/time format string.
*
* @param formatString The format string that will be given to {@code SimpleDateFormat} to determine
- * the bucket path.
+ * the bucket id.
*/
public DateTimeBucketAssigner(String formatString) {
- this.formatString = formatString;
+ this(formatString, ZoneId.systemDefault());
+ }
+
+ /**
+ * Creates a new {@code DateTimeBucketAssigner} with format string {@code "yyyy-MM-dd--HH"} using the given timezone.
+ *
+ * @param zoneId The timezone used to format {@code DateTimeFormatter} for bucket id.
+ */
+ public DateTimeBucketAssigner(ZoneId zoneId) {
+ this(DEFAULT_FORMAT_STRING, zoneId);
+ }
+
+ /**
+ * Creates a new {@code DateTimeBucketAssigner} with the given date/time format string using the given timezone.
+ *
+ * @param formatString The format string that will be given to {@code DateTimeFormatter} to determine
+ * the bucket path.
+ * @param zoneId The timezone used to format {@code DateTimeFormatter} for bucket id.
+ */
+ public DateTimeBucketAssigner(String formatString, ZoneId zoneId) {
+ this.formatString = Preconditions.checkNotNull(formatString);
+ this.zoneId = Preconditions.checkNotNull(zoneId);
}
@Override
public String getBucketId(IN element, BucketAssigner.Context context) {
- if (dateFormatter == null) {
- dateFormatter = new SimpleDateFormat(formatString);
+ if (dateTimeFormatter == null) {
+ dateTimeFormatter = DateTimeFormatter.ofPattern(formatString).withZone(zoneId);
}
- return dateFormatter.format(new Date(context.currentProcessingTime()));
+ return dateTimeFormatter.format(Instant.ofEpochMilli(context.currentProcessingTime()));
}
@Override
@@ -94,6 +119,9 @@ public class DateTimeBucketAssigner<IN> implements BucketAssigner<IN, String> {
@Override
public String toString() {
- return "DateTimeBucketAssigner{formatString='" + formatString + '\'' + '}';
+ return "DateTimeBucketAssigner{" +
+ "formatString='" + formatString + '\'' +
+ ", zoneId=" + zoneId +
+ '}';
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssignerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssignerTest.java
new file mode 100644
index 0000000..b4fac18
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/bucketassigners/DateTimeBucketAssignerTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.streaming.api.functions.sink.filesystem.bucketassigners;
+
+import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
+
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+
+import java.time.ZoneId;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link DateTimeBucketAssigner}.
+ */
+public class DateTimeBucketAssignerTest {
+ private static final long TEST_TIME_IN_MILLIS = 1533363082011L;
+
+ private static final MockedContext mockedContext = new MockedContext();
+
+ @Test
+ public void testGetBucketPathWithSpecifiedTimezone() {
+ DateTimeBucketAssigner bucketAssigner = new DateTimeBucketAssigner(ZoneId.of("America/Los_Angeles"));
+
+ assertEquals("2018-08-03--23", bucketAssigner.getBucketId(null, mockedContext));
+ }
+
+ @Test
+ public void testGetBucketPathWithSpecifiedFormatString() {
+ DateTimeBucketAssigner bucketAssigner = new DateTimeBucketAssigner("yyyy-MM-dd-HH", ZoneId.of("America/Los_Angeles"));
+
+ assertEquals("2018-08-03-23", bucketAssigner.getBucketId(null, mockedContext));
+ }
+
+ private static class MockedContext implements BucketAssigner.Context {
+ @Override
+ public long currentProcessingTime() {
+ return TEST_TIME_IN_MILLIS;
+ }
+
+ @Override
+ public long currentWatermark() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public Long timestamp() {
+ return null;
+ }
+ }
+}