Merge branch '1.8_cassandra_fix' into 'v1.8.0_dev'

cassandra plugin



See merge request !160

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,15 +48,11 @@
 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.metric.MetricConstant;
-import org.apache.flink.api.common.io.RichOutputFormat;
+import com.dtstack.flink.sql.sink.MetricOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.Counter;
-import org.apache.flink.metrics.Meter;
-import org.apache.flink.metrics.MeterView;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -73,7 +69,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 +90,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 +108,8 @@ 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();
         try {
             if (session == null) {
                 QueryOptions queryOptions = new QueryOptions();
@@ -176,17 +165,12 @@ public void open(int taskNumber, int numTasks) throws IOException {
                 // 建立连接 连接已存在的键空间
                 session = cluster.connect(database);
                 LOG.info("connect cassandra is successed!");
-                initMetric();
             }
         } catch (Exception e) {
             LOG.error("connect cassandra is error:" + e.getMessage());
         }
     }
 
-    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,6 +210,7 @@ private void insertWrite(Row row) {
                 resultSet.wasApplied();
             }
         } catch (Exception e) {
+            outDirtyRecords.inc();
             LOG.error("[upsert] is error:" + e.getMessage());
         }
     }
@@ -237,7 +222,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);