Merge branch 'v1.8.0_dev_cassandra' into 'v1.8.0_dev'

卡桑德啦 监控指标



See merge request !155

Author: toutian

cassandra/cassandra-side/cassandra-all-side/pom.xml

@@ -38,7 +38,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

cassandra/cassandra-side/cassandra-async-side/pom.xml

@@ -54,7 +54,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -174,9 +174,16 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                 return;
             }
             inputParams.add(equalObj);
-            stringBuffer.append(sideInfo.getEqualFieldList().get(i))
-                    .append(" = ").append("'" + equalObj + "'")
-                    .append(" and ");
+            StringBuffer sqlTemp = stringBuffer.append(sideInfo.getEqualFieldList().get(i))
+                    .append(" = ");
+            if (equalObj instanceof String) {
+                sqlTemp.append("'" + equalObj + "'")
+                        .append(" and ");
+            } else {
+                sqlTemp.append(equalObj)
+                        .append(" and ");
+            }
+
         }
 
         String key = buildCacheKey(inputParams);
@@ -190,12 +197,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     dealMissKey(input, resultFuture);
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
-
-                    for (Object rowArray : (List) val.getContent()) {
-                        Row row = fillData(input, rowArray);
-                        resultFuture.complete(Collections.singleton(row));
+                    List<Row> rowList = Lists.newArrayList();
+                    for (Object jsonArray : (List) val.getContent()) {
+                        Row row = fillData(input, jsonArray);
+                        rowList.add(row);
                     }
-
+                    resultFuture.complete(rowList);
                 } else {
                     throw new RuntimeException("not support cache obj type " + val.getType());
                 }
@@ -206,7 +213,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         //connect Cassandra
         connCassandraDB(cassandraSideTableInfo);
 
-        String sqlCondition = sideInfo.getSqlCondition() + " " + sqlWhere;
+        String sqlCondition = sideInfo.getSqlCondition() + " " + sqlWhere + "  ALLOW FILTERING ";
         System.out.println("sqlCondition:" + sqlCondition);
 
         ListenableFuture<ResultSet> resultSet = Futures.transformAsync(session,
@@ -231,14 +238,15 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                 cluster.closeAsync();
                 if (rows.size() > 0) {
                     List<com.datastax.driver.core.Row> cacheContent = Lists.newArrayList();
+                    List<Row> rowList = Lists.newArrayList();
                     for (com.datastax.driver.core.Row line : rows) {
                         Row row = fillData(input, line);
                         if (openCache()) {
                             cacheContent.add(line);
                         }
-                        resultFuture.complete(Collections.singleton(row));
+                        rowList.add(row);
                     }
-
+                    resultFuture.complete(rowList);
                     if (openCache()) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                     }
@@ -285,7 +293,6 @@ public Row fillData(Row input, Object line) {
             }
         }
 
-        System.out.println("row:" + row.toString());
         return row;
     }
 

cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideParser.java

@@ -23,6 +23,9 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
 
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -96,4 +99,32 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
 
     private static void dealSideSign(Matcher matcher, TableInfo tableInfo) {
     }
+
+    public Class dbTypeConvertToJavaType(String fieldType) {
+        switch (fieldType.toLowerCase()) {
+            case "bigint":
+                return Long.class;
+            case "int":
+            case "counter":
+                return Integer.class;
+
+            case "text":
+            case "inet":
+            case "varchar":
+            case "ascii":
+            case "timeuuid":
+                return String.class;
+
+            case "decimal":
+            case "float":
+                return Float.class;
+            case "double":
+                return Double.class;
+            case "timestamp":
+                return Timestamp.class;
+        }
+
+        throw new RuntimeException("不支持 " + fieldType + " 类型");
+
+    }
 }

cassandra/cassandra-sink/pom.xml

@@ -33,7 +33,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -48,7 +48,9 @@
 import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
+import com.dtstack.flink.sql.config.DirtyConfig;
 import com.dtstack.flink.sql.metric.MetricConstant;
+import com.dtstack.flink.sql.sink.MetricOutputFormat;
 import org.apache.flink.api.common.io.RichOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple;
@@ -73,7 +75,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends RichOutputFormat<Tuple2> {
+public class CassandraOutputFormat extends MetricOutputFormat {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);
@@ -94,17 +96,9 @@ public class CassandraOutputFormat extends RichOutputFormat<Tuple2> {
     protected String[] fieldNames;
     TypeInformation<?>[] fieldTypes;
 
-    private int batchInterval = 5000;
-
     private Cluster cluster;
     private Session session = null;
 
-    private int batchCount = 0;
-
-    private transient Counter outRecords;
-
-    private transient Meter outRecordsRate;
-
     public CassandraOutputFormat() {
     }
 
@@ -120,7 +114,9 @@ public void configure(Configuration parameters) {
      *                     I/O problem.
      */
     @Override
-    public void open(int taskNumber, int numTasks) throws IOException {
+    public void open(int taskNumber, int numTasks) {
+        initMetric();
+        initDirtyDataOutputStream();
         try {
             if (session == null) {
                 QueryOptions queryOptions = new QueryOptions();
@@ -183,10 +179,6 @@ public void open(int taskNumber, int numTasks) throws IOException {
         }
     }
 
-    private void initMetric() {
-        outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
-        outRecordsRate = getRuntimeContext().getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(outRecords, 20));
-    }
 
     /**
      * Adds a record to the prepared statement.
@@ -226,7 +218,7 @@ private void insertWrite(Row row) {
                 resultSet.wasApplied();
             }
         } catch (Exception e) {
-            LOG.error("[upsert] is error:" + e.getMessage());
+            dealInsertError(row.toString(), e);
         }
     }
 
@@ -237,7 +229,11 @@ private String buildSql(Row row) {
             if (row.getField(index) == null) {
             } else {
                 fields.append(fieldNames[index] + ",");
-                values.append("'" + row.getField(index) + "'" + ",");
+                if (row.getField(index) instanceof String) {
+                    values.append("'" + row.getField(index) + "'" + ",");
+                } else {
+                    values.append(row.getField(index) + ",");
+                }
             }
         }
         fields.deleteCharAt(fields.length() - 1);
@@ -352,6 +348,12 @@ public CassandraFormatBuilder setPoolTimeoutMillis(Integer poolTimeoutMillis) {
             return this;
         }
 
+        public CassandraFormatBuilder setDirtyConfig(DirtyConfig dirtyConfig) {
+            format.dirtyConfig = dirtyConfig;
+            return this;
+        }
+
+
         /**
          * Finalizes the configuration and checks validity.
          *

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraSink.java

@@ -20,6 +20,7 @@
 package com.dtstack.flink.sql.sink.cassandra;
 
 
+import com.dtstack.flink.sql.config.DirtyConfig;
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.cassandra.table.CassandraTableInfo;
 import com.dtstack.flink.sql.table.TargetTableInfo;
@@ -57,6 +58,7 @@ public class CassandraSink implements RetractStreamTableSink<Row>, IStreamSinkGe
     protected Integer readTimeoutMillis;
     protected Integer connectTimeoutMillis;
     protected Integer poolTimeoutMillis;
+    protected DirtyConfig dirtyConfig;
 
     public CassandraSink() {
         // TO DO NOTHING
@@ -77,6 +79,7 @@ public CassandraSink genStreamSink(TargetTableInfo targetTableInfo) {
         this.readTimeoutMillis = cassandraTableInfo.getReadTimeoutMillis();
         this.connectTimeoutMillis = cassandraTableInfo.getConnectTimeoutMillis();
         this.poolTimeoutMillis = cassandraTableInfo.getPoolTimeoutMillis();
+        this.dirtyConfig = cassandraTableInfo.getDirtyConfig();
         return this;
     }
 
@@ -96,7 +99,9 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 .setConnectTimeoutMillis(this.connectTimeoutMillis)
                 .setPoolTimeoutMillis(this.poolTimeoutMillis)
                 .setFieldNames(this.fieldNames)
-                .setFieldTypes(this.fieldTypes);
+                .setFieldTypes(this.fieldTypes)
+                .setDirtyConfig(dirtyConfig);
+
 
         CassandraOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);