commit 41b612d9d5fb359f6ef55c092b8f6479973fd1d3
author zhenhb <894672675@qq.com> Sat Jul 09 18:07:20 2022 +0800
committer GitHub <noreply@github.com> Sat Jul 09 18:07:20 2022 +0800
tree 57f527896eafc0044065692028bc40c5f430dcf6
parent 2e38c12ad80f9647e07547875045afaf386f74dc

[Enhancement] Add param max_filter_ratio for sink of spark connector (#41)

spark-doris-connector/src/main/java/org/apache/doris/spark/DorisStreamLoad.java

diff --git a/spark-doris-connector/src/main/java/org/apache/doris/spark/DorisStreamLoad.java b/spark-doris-connector/src/main/java/org/apache/doris/spark/DorisStreamLoad.java
index db3ef7d..be3c938 100644
--- a/spark-doris-connector/src/main/java/org/apache/doris/spark/DorisStreamLoad.java
+++ b/spark-doris-connector/src/main/java/org/apache/doris/spark/DorisStreamLoad.java
@@ -68,6 +68,7 @@
     private String authEncoding;
     private String columns;
     private String[] dfColumns;
+    private String maxFilterRatio;
 
     public DorisStreamLoad(String hostPort, String db, String tbl, String user, String passwd) {
         this.hostPort = hostPort;
@@ -90,6 +91,9 @@
         this.loadUrlStr = String.format(loadUrlPattern, hostPort, db, tbl);
         this.authEncoding = Base64.getEncoder().encodeToString(String.format("%s:%s", user, passwd).getBytes(StandardCharsets.UTF_8));
         this.columns = settings.getProperty(ConfigurationOptions.DORIS_WRITE_FIELDS);
+
+        this.maxFilterRatio = settings.getProperty(ConfigurationOptions.DORIS_MAX_FILTER_RATIO);
+
     }
 
     public DorisStreamLoad(SparkSettings settings, String[] dfColumns) throws IOException, DorisException {
@@ -100,10 +104,14 @@
         this.tbl = dbTable[1];
         this.user = settings.getProperty(ConfigurationOptions.DORIS_REQUEST_AUTH_USER);
         this.passwd = settings.getProperty(ConfigurationOptions.DORIS_REQUEST_AUTH_PASSWORD);
+
+
         this.loadUrlStr = String.format(loadUrlPattern, hostPort, db, tbl);
         this.authEncoding = Base64.getEncoder().encodeToString(String.format("%s:%s", user, passwd).getBytes(StandardCharsets.UTF_8));
         this.columns = settings.getProperty(ConfigurationOptions.DORIS_WRITE_FIELDS);
         this.dfColumns = dfColumns;
+
+        this.maxFilterRatio = settings.getProperty(ConfigurationOptions.DORIS_MAX_FILTER_RATIO);
     }
 
     public String getLoadUrlStr() {
@@ -132,6 +140,11 @@
         if (columns != null && !columns.equals("")) {
             conn.addRequestProperty("columns", columns);
         }
+
+        if (maxFilterRatio != null && !maxFilterRatio.equals("")) {
+            conn.addRequestProperty("max_filter_ratio", maxFilterRatio);
+        }
+
         conn.setDoOutput(true);
         conn.setDoInput(true);
         conn.addRequestProperty("format", "json");

spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java

diff --git a/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java b/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
index 9a0cead..e3c55d6 100644
--- a/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
+++ b/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
@@ -71,4 +71,6 @@
 
     String DORIS_SINK_MAX_RETRIES = "doris.sink.max-retries";
     int SINK_MAX_RETRIES_DEFAULT = 1;
+
+    String DORIS_MAX_FILTER_RATIO = "doris.max.filter.ratio";
 }