This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-spark-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 95c278d [refactor](stream-load) Adjust the content of the review
Added direct load via benodes (#177)
95c278d is described below
commit 95c278d6e5e5d47e8b3b97161a1c8c9b54c18a64
Author: MYiYang <[email protected]>
AuthorDate: Thu Jan 4 14:20:16 2024 +0800
[refactor](stream-load) Adjust the content of the review Added direct load
via benodes (#177)
---
.../doris/spark/cfg/ConfigurationOptions.java | 7 +-
.../apache/doris/spark/load/DorisStreamLoad.java | 8 +-
.../org/apache/doris/spark/rest/RestService.java | 147 ++++++++++++++++-----
.../apache/doris/spark/rest/models/BackendV2.java | 16 ++-
4 files changed, 140 insertions(+), 38 deletions(-)
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 7877bc8..64cb38f 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
@@ -20,6 +20,8 @@ package org.apache.doris.spark.cfg;
public interface ConfigurationOptions {
// doris fe node address
String DORIS_FENODES = "doris.fenodes";
+
+ String DORIS_BENODES = "doris.benodes";
String DORIS_QUERY_PORT = "doris.query.port";
String DORIS_DEFAULT_CLUSTER = "default_cluster";
@@ -124,4 +126,7 @@ public interface ConfigurationOptions {
*/
String DORIS_SINK_AUTO_REDIRECT = "doris.sink.auto-redirect";
boolean DORIS_SINK_AUTO_REDIRECT_DEFAULT = false;
-}
+
+
+
+}
\ No newline at end of file
diff --git
a/spark-doris-connector/src/main/java/org/apache/doris/spark/load/DorisStreamLoad.java
b/spark-doris-connector/src/main/java/org/apache/doris/spark/load/DorisStreamLoad.java
index 7758c6f..9fdf4c8 100644
---
a/spark-doris-connector/src/main/java/org/apache/doris/spark/load/DorisStreamLoad.java
+++
b/spark-doris-connector/src/main/java/org/apache/doris/spark/load/DorisStreamLoad.java
@@ -98,6 +98,7 @@ public class DorisStreamLoad implements Serializable {
private static final long cacheExpireTimeout = 4 * 60;
private final LoadingCache<String, List<BackendV2.BackendRowV2>> cache;
private final String fenodes;
+ private final String benodes;
private final DataFormat dataFormat;
private String FIELD_DELIMITER;
private final String LINE_DELIMITER;
@@ -112,6 +113,7 @@ public class DorisStreamLoad implements Serializable {
this.db = dbTable[0];
this.tbl = dbTable[1];
this.fenodes =
settings.getProperty(ConfigurationOptions.DORIS_FENODES);
+ this.benodes =
settings.getProperty(ConfigurationOptions.DORIS_BENODES);
String user =
settings.getProperty(ConfigurationOptions.DORIS_REQUEST_AUTH_USER);
String passwd =
settings.getProperty(ConfigurationOptions.DORIS_REQUEST_AUTH_PASSWORD);
this.authEncoded = getAuthEncoded(user, passwd);
@@ -421,7 +423,8 @@ public class DorisStreamLoad implements Serializable {
return backend.getIp() + ":" + backend.getHttpPort();
} catch (ExecutionException e) {
throw new RuntimeException("get backends info fail", e);
- } catch (IllegalArgumentException e) {
+ }
+ catch (IllegalArgumentException e) {
throw new RuntimeException("get frontend info fail", e);
}
}
@@ -444,9 +447,8 @@ public class DorisStreamLoad implements Serializable {
@Override
public List<BackendV2.BackendRowV2> load(String key) throws Exception {
- return RestService.getBackendRows(settings, LOG);
+ return RestService.getBackendRows(settings, LOG);
}
-
}
private String generateLoadLabel() {
diff --git
a/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/RestService.java
b/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/RestService.java
index 2140c02..86e370c 100644
---
a/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/RestService.java
+++
b/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/RestService.java
@@ -26,6 +26,7 @@ import static
org.apache.doris.spark.cfg.ConfigurationOptions.DORIS_TABLET_SIZE;
import static
org.apache.doris.spark.cfg.ConfigurationOptions.DORIS_TABLET_SIZE_DEFAULT;
import static
org.apache.doris.spark.cfg.ConfigurationOptions.DORIS_TABLET_SIZE_MIN;
import static
org.apache.doris.spark.cfg.ConfigurationOptions.DORIS_TABLE_IDENTIFIER;
+import static org.apache.doris.spark.cfg.ConfigurationOptions.DORIS_BENODES;
import static org.apache.doris.spark.util.ErrorMessages.CONNECT_FAILED_MESSAGE;
import static
org.apache.doris.spark.util.ErrorMessages.ILLEGAL_ARGUMENT_MESSAGE;
import static
org.apache.doris.spark.util.ErrorMessages.PARSE_NUMBER_FAILED_MESSAGE;
@@ -517,10 +518,31 @@ public class RestService implements Serializable {
@Deprecated
@VisibleForTesting
public static String randomBackend(SparkSettings sparkSettings , Logger
logger) throws DorisException {
+ return getBackend(sparkSettings, logger);
+ }
+
+ @Deprecated
+ @VisibleForTesting
+ public static String beBackend(SparkSettings sparkSettings , Logger
logger) throws DorisException {
+ return getBackend(sparkSettings, logger);
+ }
+
+ /**
+ * choice a Doris BE node to request.
+ * @param logger slf4j logger
+ * @return the chosen one Doris BE node
+ * @throws IllegalArgumentException BE nodes is illegal
+ */
+ @VisibleForTesting
+ public static String randomBackendV2(SparkSettings sparkSettings, Logger
logger) throws DorisException {
+ return getBackend(sparkSettings, logger);
+ }
+
+ private static String getBackend(SparkSettings sparkSettings, Logger
logger) throws DorisException {
List<BackendV2.BackendRowV2> backends = getBackendRows(sparkSettings,
logger);
Collections.shuffle(backends);
BackendV2.BackendRowV2 backend = backends.get(0);
- return backend.getIp()+ ":" + backend.getHttpPort();
+ return backend.getIp() + ":" + backend.getHttpPort();
}
/**
@@ -561,48 +583,72 @@ public class RestService implements Serializable {
}
/**
- * get Doris BE node list.
+ * get Doris BE nodes.
* @param logger slf4j logger
- * @return the Doris BE node list
+ * @return the Doris BE node
* @throws IllegalArgumentException BE nodes is illegal
*/
- @VisibleForTesting
- public static List<BackendV2.BackendRowV2> getBackendRows(SparkSettings
sparkSettings, Logger logger) throws DorisException {
- List<String> feNodeList =
allEndpoints(sparkSettings.getProperty(DORIS_FENODES), logger);
- for (String feNode : feNodeList){
+ public static List<BackendV2.BackendRowV2> getBeNodes(SparkSettings
sparkSettings, Logger logger) throws DorisException {
+ List<String> backends =
allBeEndpoints(sparkSettings.getProperty(DORIS_BENODES),logger);
+ List<BackendV2.BackendRowV2> backendRowV2s = new
ArrayList<BackendV2.BackendRowV2>();
+ if (backends == null || backends.isEmpty()) {
+ logger.error(ILLEGAL_ARGUMENT_MESSAGE, "benodes", backends);
+ throw new IllegalArgumentException("benodes",
String.valueOf(backends));
+ }
+ BackendV2.BackendRowV2 backendRowV2 = new BackendV2.BackendRowV2();
+ for (int i = 0; i < backends.size(); i++) {
+ String ip =
backends.get(i).substring(0,backends.get(i).indexOf(":"));
try {
- String beUrl = String.format("http://%s"; + BACKENDS_V2,
feNode);
- HttpGet httpGet = new HttpGet(beUrl);
- String response = send(sparkSettings, httpGet, logger);
- logger.info("Backend Info:{}", response);
- List<BackendV2.BackendRowV2> backends =
parseBackendV2(response, logger);
- logger.trace("Parse beNodes '{}'.", backends);
- if (backends == null || backends.isEmpty()) {
- logger.error(ILLEGAL_ARGUMENT_MESSAGE, "beNodes",
backends);
- throw new IllegalArgumentException("beNodes",
String.valueOf(backends));
- }
- return backends;
- } catch (ConnectedFailedException e) {
- logger.info("Doris FE node {} is unavailable: {}, Request the
next Doris FE node", feNode, e.getMessage());
+ Integer port =
Integer.valueOf(backends.get(i).substring(backends.get(i).indexOf(":")+1,backends.get(i).length()));
+ /**
+ * By default, the BE port you enter is is_alive=true
+ */
+ BackendV2.BackendRowV2 backend =
backendRowV2.of(ip,port,true);
+ backendRowV2s.add(backend);
+ } catch (NumberFormatException e) {
+ logger.error("Doris BE is port error, please check
configuration");
+ throw new RuntimeException(e);
}
}
- String errMsg = "No Doris FE is available, please check configuration";
- logger.error(errMsg);
- throw new DorisException(errMsg);
+ return backendRowV2s;
}
/**
- * choice a Doris BE node to request.
+ * get Doris BE node list.
* @param logger slf4j logger
- * @return the chosen one Doris BE node
+ * @return the Doris BE node list
* @throws IllegalArgumentException BE nodes is illegal
*/
@VisibleForTesting
- public static String randomBackendV2(SparkSettings sparkSettings, Logger
logger) throws DorisException {
- List<BackendV2.BackendRowV2> backends = getBackendRows(sparkSettings,
logger);
- Collections.shuffle(backends);
- BackendV2.BackendRowV2 backend = backends.get(0);
- return backend.getIp() + ":" + backend.getHttpPort();
+ public static List<BackendV2.BackendRowV2> getBackendRows(SparkSettings
sparkSettings, Logger logger) throws DorisException {
+ /**
+ * If the specified BE does not exist, the FE mode is used
+ */
+ if(notBeNode(sparkSettings,logger)){
+ List<String> feNodeList =
allEndpoints(sparkSettings.getProperty(DORIS_FENODES), logger);
+ for (String feNode : feNodeList){
+ try {
+ String beUrl = String.format("http://%s"; + BACKENDS_V2,
feNode);
+ HttpGet httpGet = new HttpGet(beUrl);
+ String response = send(sparkSettings, httpGet, logger);
+ logger.info("Backend Info:{}", response);
+ List<BackendV2.BackendRowV2> backends =
parseBackendV2(response, logger);
+ logger.trace("Parse benodes '{}'.", backends);
+ if (backends == null || backends.isEmpty()) {
+ logger.error(ILLEGAL_ARGUMENT_MESSAGE, "benodes",
backends);
+ throw new IllegalArgumentException("benodes",
String.valueOf(backends));
+ }
+ return backends;
+ } catch (ConnectedFailedException e) {
+ logger.info("Doris FE node {} is unavailable: {}, Request
the next Doris FE node", feNode, e.getMessage());
+ }
+ }
+ String errMsg = "No Doris FE is available, please check
configuration";
+ logger.error(errMsg);
+ throw new DorisException(errMsg);
+ }else {
+ return getBeNodes(sparkSettings, logger);
+ }
}
static List<BackendV2.BackendRowV2> parseBackendV2(String response, Logger
logger) throws DorisException {
@@ -646,7 +692,7 @@ public class RestService implements Serializable {
*/
@VisibleForTesting
static List<PartitionDefinition> tabletsMapToPartition(Settings cfg,
Map<String, List<Long>> be2Tablets,
- String opaquedQueryPlan, String database, String table, Logger
logger)
+ String
opaquedQueryPlan, String database, String table, Logger logger)
throws IllegalArgumentException {
int tabletsSize = tabletCountLimitForOnePartition(cfg, logger);
List<PartitionDefinition> partitions = new ArrayList<>();
@@ -689,4 +735,41 @@ public class RestService implements Serializable {
Collections.shuffle(nodes);
return nodes;
}
-}
+
+ /**
+ * choice a Doris BE node to request.
+ *
+ * @param beNodes Doris BE node list, separate be comma
+ * @param logger slf4j logger
+ * @return the array of Doris FE nodes
+ * @throws IllegalArgumentException fe nodes is illegal
+ */
+ @VisibleForTesting
+ static List<String> allBeEndpoints(String beNodes, Logger logger) throws
IllegalArgumentException {
+ logger.trace("Parse benodes '{}'.", beNodes);
+ if (StringUtils.isEmpty(beNodes)) {
+ logger.error(ILLEGAL_ARGUMENT_MESSAGE, "benodes", beNodes);
+ throw new IllegalArgumentException("benodes", beNodes);
+ }
+ List<String> nodes =
Arrays.stream(beNodes.split(",")).map(String::trim).collect(Collectors.toList());
+ Collections.shuffle(nodes);
+ return nodes;
+ }
+
+
+ /**
+ * Doris BE node is not
+ * @param logger slf4j logger
+ * @return Doris BE node Yes or no
+ */
+ public static Boolean notBeNode(SparkSettings sparkSettings, Logger
logger){
+ String beNodes = sparkSettings.getProperty(DORIS_BENODES);
+ if(null == beNodes){
+ return true;
+ }else {
+ return false;
+ }
+ }
+
+
+}
\ No newline at end of file
diff --git
a/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/models/BackendV2.java
b/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/models/BackendV2.java
index 75a2514..858408a 100644
---
a/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/models/BackendV2.java
+++
b/spark-doris-connector/src/main/java/org/apache/doris/spark/rest/models/BackendV2.java
@@ -26,6 +26,7 @@ import java.util.List;
@JsonIgnoreProperties(ignoreUnknown = true)
public class BackendV2 {
+
@JsonProperty(value = "backends")
private List<BackendRowV2> backends;
@@ -37,7 +38,11 @@ public class BackendV2 {
this.backends = rows;
}
- public static class BackendRowV2 {
+ public static class BackendRowV2 implements Cloneable{
+
+ public BackendRowV2(){
+ super();
+ }
@JsonProperty("ip")
public String ip;
@JsonProperty("http_port")
@@ -68,5 +73,12 @@ public class BackendV2 {
public void setAlive(boolean alive) {
isAlive = alive;
}
+ public static BackendRowV2 of(String ip, int httpPort, boolean alive) {
+ BackendRowV2 rowV2 = new BackendRowV2();
+ rowV2.setIp(ip);
+ rowV2.setHttpPort(httpPort);
+ rowV2.setAlive(alive);
+ return rowV2;
+ }
}
-}
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
