自定义DateTimeBucket

转载自&＃xff1a;https://blog.csdn.net/u010259977/article/details/88683503

用flink消费kafka内容&＃xff0c;通过清洗、转换、过滤后&＃xff0c;要sink到parquet文件&＃xff0c;需要按照事件的event进行分区生产需要写入的文件夹&＃xff0c;如event1的发生时间在2018-03-19&＃xff0c;而event2的发生时间在2018-03-20&＃xff0c;这就涉及到extract它的eventtime&＃xff0c;并生产parquet文件的bucktId,具体代码如下&＃xff1a;

1. /*

2. * Licensed to the Apache Software Foundation (ASF) under one

3. * or more contributor license agreements. See the NOTICE file

4. * distributed with this work for additional information

5. * regarding copyright ownership. The ASF licenses this file

6. * to you under the Apache License, Version 2.0 (the

7. * "License"); you may not use this file except in compliance

8. * with the License. You may obtain a copy of the License at

9. *

10. * http://www.apache.org/licenses/LICENSE-2.0

11. *

12. * Unless required by applicable law or agreed to in writing, software

13. * distributed under the License is distributed on an "AS IS" BASIS,

14. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

15. * See the License for the specific language governing permissions and

16. * limitations under the License.

17. */

18.  

19. package com.hellobike.realtimeplatform.utils;

20.  

21. import com.hellobike.realtimeplatform.model.AccessLog;

22. import org.apache.flink.annotation.PublicEvolving;

23. import org.apache.flink.core.io.SimpleVersionedSerializer;

24. import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;

25. import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;

26. import org.apache.flink.util.Preconditions;

27.  

28. import java.text.ParseException;

29. import java.text.SimpleDateFormat;

30. import java.time.Instant;

31. import java.time.ZoneId;

32. import java.time.format.DateTimeFormatter;

33. import java.util.Calendar;

34.  

35. /**

36. * A {&＃64;link BucketAssigner} that assigns to buckets based on current system time.

37. *

38. *

39. *

    The {&＃64;code DateTimeBucketer} will create directories of the following form:

40. * {&＃64;code /{basePath}/{dateTimePath}/}. The {&＃64;code basePath} is the path

41. * that was specified as a base path when creating the

42. * {&＃64;link org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink}.

43. * The {&＃64;code dateTimePath} is determined based on the current system time and the

44. * user provided format string.

45. *

46. *

47. *

    {&＃64;link DateTimeFormatter} is used to derive a date string from the current system time and

48. * the date format string. The default format string is {&＃64;code "yyyy-MM-dd--HH"} so the rolling

49. * files will have a granularity of hours.

50. *

51. *

    Example:

52. *

53. *

    {&＃64;code
     

54. * BucketAssigner bucketAssigner &＃61; new DateTimeBucketAssigner("yyyy-MM-dd--HH");

55. * }

56. *

57. *

    This will create for example the following bucket path:

58. * {&＃64;code /base/1976-12-31-14/}

59. */

60. &＃64;PublicEvolving

61. public class DateTimeBucketWithPartitionAssigner implements BucketAssigner {

62.  

63. private static final long serialVersionUID &＃61; 1L;

64.  

65. private static final String DEFAULT_FORMAT_STRING &＃61; "yyyy-MM-dd--HH";

66.  

67. private final String formatString;

68.  

69. private final ZoneId zoneId;

70.  

71. private transient DateTimeFormatter dateTimeFormatter;

72.  

73. /**

74. * Creates a new {&＃64;code DateTimeBucketAssigner} with format string {&＃64;code "yyyy-MM-dd--HH"}.

75. */

76. public DateTimeBucketWithPartitionAssigner() {

77. this(DEFAULT_FORMAT_STRING);

78. }

79.  

80. /**

81. * Creates a new {&＃64;code DateTimeBucketAssigner} with the given date/time format string.

82. *

83. * &＃64;param formatString The format string that will be given to {&＃64;code SimpleDateFormat} to determine

84. * the bucket id.

85. */

86. public DateTimeBucketWithPartitionAssigner(String formatString) {

87. this(formatString, ZoneId.systemDefault());

88. }

89.  

90. /**

91. * Creates a new {&＃64;code DateTimeBucketAssigner} with format string {&＃64;code "yyyy-MM-dd--HH"} using the given timezone.

92. *

93. * &＃64;param zoneId The timezone used to format {&＃64;code DateTimeFormatter} for bucket id.

94. */

95. public DateTimeBucketWithPartitionAssigner(ZoneId zoneId) {

96. this(DEFAULT_FORMAT_STRING, zoneId);

97. }

98.  

99. /**

100. * Creates a new {&＃64;code DateTimeBucketAssigner} with the given date/time format string using the given timezone.

101. *

102. * &＃64;param formatString The format string that will be given to {&＃64;code DateTimeFormatter} to determine

103. * the bucket path.

104. * &＃64;param zoneId The timezone used to format {&＃64;code DateTimeFormatter} for bucket id.

105. */

106. public DateTimeBucketWithPartitionAssigner(String formatString, ZoneId zoneId) {

107. this.formatString &＃61; Preconditions.checkNotNull(formatString);

108. this.zoneId &＃61; Preconditions.checkNotNull(zoneId);

109. }

110.  

111. &＃64;Override

112. public String getBucketId(IN element, Context context) {

113. long eventTime &＃61; 0L;

114. if (dateTimeFormatter &＃61;&＃61; null) {

115. dateTimeFormatter &＃61; DateTimeFormatter.ofPattern(formatString).withZone(zoneId);

116. }

117. if (element instanceof AccessLog) {

118. if (-1 &＃61;&＃61; ((AccessLog) element).getParseResult() && "".equals(((AccessLog) element).getEventTimestamp())) {

119. return "pt&＃61;errorTime";

120. }

121.  

122. eventTime &＃61; Long.valueOf(((AccessLog) element).getEventTimestamp());

123.  

124. return "pt&＃61;" &＃43; dateTimeFormatter.format(Instant.ofEpochMilli(eventTime));

125. } else {

126. return "pt&＃61;errorObjects";

127. }

128. }

129.  
130.  

131. &＃64;Override

132. public SimpleVersionedSerializer getSerializer() {

133. return SimpleVersionedStringSerializer.INSTANCE;

134. }

135.  

136. &＃64;Override

137. public String toString() {

138. return "DateTimeBucketAssigner{" &＃43;

139. "formatString&＃61;&＃39;" &＃43; formatString &＃43; &＃39;\&＃39;&＃39; &＃43;

140. ", zoneId&＃61;" &＃43; zoneId &＃43;

141. &＃39;}&＃39;;

142. }

143. }

指定自定义的DateTime Assigner就可以实现基于event time分桶写入parquet文件