Skip to content

Commit 57b9d28

Browse files
zoudaokoulifezoudaokoulife
committed
Merge branch 'v1.5.0_dev_parse_json_dataformat' into 'v1.5.0_dev'
 kafka解析timestamp类型并打印脏数据日志 See merge request !58
2 parents 7d2c65e + 0548b64 commit 57b9d28

File tree

4 files changed

+123
-19
lines changed

4 files changed

+123
-19
lines changed

‎kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java‎

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,12 @@
2424
import com.dtstack.flink.sql.source.AbsDeserialization;
2525
import com.dtstack.flink.sql.source.kafka.metric.KafkaTopicPartitionLagMetric;
2626
import org.apache.flink.api.common.typeinfo.TypeInformation;
27+
import org.apache.flink.api.common.typeinfo.Types;
2728
import org.apache.flink.api.java.typeutils.RowTypeInfo;
2829
import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
2930
import org.apache.flink.metrics.MetricGroup;
3031
import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
32+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
3133
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
3234
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
3335
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeType;
@@ -43,6 +45,9 @@
4345

4446
import java.io.IOException;
4547
import java.lang.reflect.Field;
48+
import java.sql.Date;
49+
import java.sql.Time;
50+
import java.sql.Timestamp;
4651
import java.util.Iterator;
4752
import java.util.Map;
4853
import java.util.Set;
@@ -62,7 +67,7 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
6267

6368
private static final long serialVersionUID = 2385115520960444192L;
6469

65-
private static int rowLenth = 1000;
70+
private static int dirtyDataFrequency = 1000;
6671

6772
private final ObjectMapper objectMapper = new ObjectMapper();
6873

@@ -112,7 +117,7 @@ public Row deserialize(byte[] message) throws IOException {
112117

113118
try {
114119
JsonNode root = objectMapper.readTree(message);
115-
if (numInRecord.getCount()%rowLenth == 0){
120+
if (numInRecord.getCount() % dirtyDataFrequency == 0) {
116121
LOG.info(root.toString());
117122
}
118123

@@ -134,17 +139,18 @@ public Row deserialize(byte[] message) throws IOException {
134139
}
135140
} else {
136141
// Read the value as specified type
137-
Object value = objectMapper.treeToValue(node, fieldTypes[i].getTypeClass());
142+
Object value = convert(node, fieldTypes[i]);
138143
row.setField(i, value);
139144
}
140145
}
141146

142147
numInResolveRecord.inc();
143148
return row;
144-
} catch (Throwable t) {
149+
} catch (Exception e) {
145150
//add metric of dirty data
146-
if (dirtyDataCounter.getCount()%rowLenth == 0){
151+
if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
147152
LOG.info("dirtyData: " + new String(message));
153+
LOG.info(" " ,e);
148154
}
149155
dirtyDataCounter.inc();
150156
return null;
@@ -243,4 +249,31 @@ protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exceptio
243249
private static String partitionLagMetricName(TopicPartition tp) {
244250
return tp + ".records-lag";
245251
}
252+
253+
private Object convert(JsonNode node, TypeInformation<?> info) {
254+
if (info.getTypeClass().equals(Types.BOOLEAN.getTypeClass())) {
255+
return node.asBoolean();
256+
} else if (info.getTypeClass().equals(Types.STRING.getTypeClass())) {
257+
return node.asText();
258+
} else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
259+
return Date.valueOf(node.asText());
260+
} else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
261+
// local zone
262+
return Time.valueOf(node.asText());
263+
} else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
264+
// local zone
265+
return Timestamp.valueOf(node.asText());
266+
} else {
267+
// for types that were specified without JSON schema
268+
// e.g. POJOs
269+
try {
270+
return objectMapper.treeToValue(node, info.getTypeClass());
271+
} catch (JsonProcessingException e) {
272+
throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
273+
}
274+
}
275+
}
276+
277+
278+
246279
}

‎kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java‎

Lines changed: 35 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,12 @@
2424
import com.dtstack.flink.sql.source.AbsDeserialization;
2525
import com.dtstack.flink.sql.source.kafka.metric.KafkaTopicPartitionLagMetric;
2626
import org.apache.flink.api.common.typeinfo.TypeInformation;
27+
import org.apache.flink.api.common.typeinfo.Types;
2728
import org.apache.flink.api.java.typeutils.RowTypeInfo;
2829
import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
2930
import org.apache.flink.metrics.MetricGroup;
3031
import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
32+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
3133
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
3234
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
3335
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeType;
@@ -43,6 +45,9 @@
4345

4446
import java.io.IOException;
4547
import java.lang.reflect.Field;
48+
import java.sql.Date;
49+
import java.sql.Time;
50+
import java.sql.Timestamp;
4651
import java.util.Iterator;
4752
import java.util.Map;
4853
import java.util.Set;
@@ -62,7 +67,7 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
6267

6368
private static final long serialVersionUID = 2385115520960444192L;
6469

65-
private static int rowLenth = 1000;
70+
private static int dirtyDataFrequency = 1000;
6671

6772
private final ObjectMapper objectMapper = new ObjectMapper();
6873

@@ -112,7 +117,7 @@ public Row deserialize(byte[] message) throws IOException {
112117
try {
113118
JsonNode root = objectMapper.readTree(message);
114119

115-
if (numInRecord.getCount()%rowLenth == 0){
120+
if (numInRecord.getCount() % dirtyDataFrequency == 0) {
116121
LOG.info(root.toString());
117122
}
118123

@@ -134,17 +139,18 @@ public Row deserialize(byte[] message) throws IOException {
134139
}
135140
} else {
136141
// Read the value as specified type
137-
Object value = objectMapper.treeToValue(node, fieldTypes[i].getTypeClass());
142+
Object value = convert(node, fieldTypes[i]);
138143
row.setField(i, value);
139144
}
140145
}
141146

142147
numInResolveRecord.inc();
143148
return row;
144-
} catch (Throwable t) {
149+
} catch (Exception e) {
145150
//add metric of dirty data
146-
if (dirtyDataCounter.getCount()%rowLenth == 0){
151+
if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
147152
LOG.info("dirtyData: " + new String(message));
153+
LOG.error(" ", e);
148154
}
149155
dirtyDataCounter.inc();
150156
return null;
@@ -244,4 +250,28 @@ protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exceptio
244250
private static String partitionLagMetricName(TopicPartition tp) {
245251
return tp + ".records-lag";
246252
}
253+
254+
private Object convert(JsonNode node, TypeInformation<?> info) {
255+
if (info.getTypeClass().equals(Types.BOOLEAN.getTypeClass())) {
256+
return node.asBoolean();
257+
} else if (info.getTypeClass().equals(Types.STRING.getTypeClass())) {
258+
return node.asText();
259+
} else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
260+
return Date.valueOf(node.asText());
261+
} else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
262+
// local zone
263+
return Time.valueOf(node.asText());
264+
} else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
265+
// local zone
266+
return Timestamp.valueOf(node.asText());
267+
} else {
268+
// for types that were specified without JSON schema
269+
// e.g. POJOs
270+
try {
271+
return objectMapper.treeToValue(node, info.getTypeClass());
272+
} catch (JsonProcessingException e) {
273+
throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
274+
}
275+
}
276+
}
247277
}

‎kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java‎

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,12 @@
2424
import com.dtstack.flink.sql.source.AbsDeserialization;
2525
import com.dtstack.flink.sql.source.kafka.metric.KafkaTopicPartitionLagMetric;
2626
import org.apache.flink.api.common.typeinfo.TypeInformation;
27+
import org.apache.flink.api.common.typeinfo.Types;
2728
import org.apache.flink.api.java.typeutils.RowTypeInfo;
2829
import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
2930
import org.apache.flink.metrics.MetricGroup;
3031
import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
32+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
3133
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
3234
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
3335
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeType;
@@ -42,7 +44,11 @@
4244
import org.slf4j.LoggerFactory;
4345

4446
import java.io.IOException;
47+
import java.lang.reflect.Array;
4548
import java.lang.reflect.Field;
49+
import java.sql.Date;
50+
import java.sql.Time;
51+
import java.sql.Timestamp;
4652
import java.util.Iterator;
4753
import java.util.Map;
4854
import java.util.Set;
@@ -64,7 +70,7 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
6470

6571
private static final long serialVersionUID = 2385115520960444192L;
6672

67-
private static int rowLenth = 1000;
73+
private static int dirtyDataFrequency = 1000;
6874

6975
private final ObjectMapper objectMapper = new ObjectMapper();
7076

@@ -115,7 +121,7 @@ public Row deserialize(byte[] message) throws IOException {
115121
try {
116122
JsonNode root = objectMapper.readTree(message);
117123

118-
if (numInRecord.getCount()%rowLenth == 0){
124+
if (numInRecord.getCount() % dirtyDataFrequency == 0) {
119125
LOG.info(root.toString());
120126
}
121127

@@ -137,17 +143,19 @@ public Row deserialize(byte[] message) throws IOException {
137143
}
138144
} else {
139145
// Read the value as specified type
140-
Object value = objectMapper.treeToValue(node, fieldTypes[i].getTypeClass());
146+
147+
Object value = convert(node, fieldTypes[i]);
141148
row.setField(i, value);
142149
}
143150
}
144151

145152
numInResolveRecord.inc();
146153
return row;
147-
} catch (Throwable t) {
154+
} catch (Exception e) {
148155
//add metric of dirty data
149-
if (dirtyDataCounter.getCount()%rowLenth == 0){
156+
if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
150157
LOG.info("dirtyData: " + new String(message));
158+
LOG.error("" , e);
151159
}
152160
dirtyDataCounter.inc();
153161
return null;
@@ -245,4 +253,29 @@ protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exceptio
245253
private static String partitionLagMetricName(TopicPartition tp) {
246254
return tp + ".records-lag";
247255
}
256+
257+
private Object convert(JsonNode node, TypeInformation<?> info) {
258+
if (info.getTypeClass().equals(Types.BOOLEAN.getTypeClass())) {
259+
return node.asBoolean();
260+
} else if (info.getTypeClass().equals(Types.STRING.getTypeClass())) {
261+
return node.asText();
262+
} else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
263+
return Date.valueOf(node.asText());
264+
} else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
265+
// local zone
266+
return Time.valueOf(node.asText());
267+
} else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
268+
// local zone
269+
return Timestamp.valueOf(node.asText());
270+
} else {
271+
// for types that were specified without JSON schema
272+
// e.g. POJOs
273+
try {
274+
return objectMapper.treeToValue(node, info.getTypeClass());
275+
} catch (JsonProcessingException e) {
276+
throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
277+
}
278+
}
279+
}
280+
248281
}

‎rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,10 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
5353

5454
private static final Logger LOG = LoggerFactory.getLogger(RetractJDBCOutputFormat.class);
5555

56+
private static int dirtyDataPrintFrequency = 1000;
57+
58+
private static int receiveDataPrintFrequency = 1000;
59+
5660
private String username;
5761
private String password;
5862
private String drivername;
@@ -167,8 +171,11 @@ public void writeRecord(Tuple2 tuple2) {
167171
}
168172

169173
if (retract) {
170-
insertWrite(row);
171174
outRecords.inc();
175+
if (outRecords.getCount() % receiveDataPrintFrequency == 0) {
176+
LOG.info("Receive data : {}", row);
177+
}
178+
insertWrite(row);
172179
} else {
173180
//do nothing
174181
}
@@ -201,8 +208,10 @@ private void writeSingleRecord(Row row) {
201208
dbConn.commit();
202209
} catch (SQLException e) {
203210
outDirtyRecords.inc();
204-
LOG.error("record insert failed ..", row.toString());
205-
LOG.error("", e);
211+
if (outDirtyRecords.getCount() % dirtyDataPrintFrequency == 0) {
212+
LOG.error("record insert failed ..", row.toString());
213+
LOG.error("", e);
214+
}
206215
}
207216
}
208217

@@ -298,7 +307,6 @@ private void updatePreparedStmt(Row row, PreparedStatement pstmt) throws SQLExce
298307

299308
private synchronized void submitExecuteBatch() {
300309
try {
301-
LOG.info("submitExecuteBatch start......");
302310
this.upload.executeBatch();
303311
dbConn.commit();
304312
} catch (SQLException e) {

0 commit comments

Comments
 (0)