cherry pick master feature (#181)

* support job parallelism for both tags and edges (#165)

* do not repartition when source dataframe has the same partition number with write partition config (#173)

* Revert "fix escapes (#152)" (#178)

This reverts commit 1f40692.

* update the numPartitions for maxcompute to Long

* update the repartition for streaming data

* support job parallelism for both tags and edges (#165)

* do not repartition when source dataframe has the same partition number with write partition config (#173)

* update the config for numPartitions

exchange-common/src/main/scala/com/vesoft/exchange/common/config/Configs.scala

@@ -172,7 +172,8 @@ case class SslConfigEntry(enableGraph: Boolean,
  }
  }
 
- override def toString: String = s"SslConfigEntry:{enableGraph:$enableGraph, enableMeta:$enableMeta, signType:${signType.toString}}"
+ override def toString: String =
+ s"SslConfigEntry:{enableGraph:$enableGraph, enableMeta:$enableMeta, signType:${signType.toString}}"
 }
 
 case class CaSignParam(caCrtFilePath: String, crtFilePath: String, keyFilePath: String)
@@ -902,7 +903,7 @@ object Configs {
  case SourceCategory.MAXCOMPUTE => {
  val table = config.getString("table")
  val partitionSpec = getStringOrNull(config, "partitionSpec")
- val numPartitions = getOrElse(config, "numPartitions", "1")
+ val numPartitions = getStringOrElse(config, "numPartitions", "1")
  val sentence = getStringOrNull(config, "sentence")
 
  MaxComputeConfigEntry(
@@ -919,7 +920,7 @@ object Configs {
  )
  }
  case SourceCategory.CLICKHOUSE => {
- val partition: String = getOrElse(config, "numPartition", "1")
+ val partition: String = getStringOrElse(config, "numPartition", "1")
  ClickHouseConfigEntry(
  SourceCategory.CLICKHOUSE,
  config.getString("url"),
@@ -1007,6 +1008,22 @@ object Configs {
  }
  }
 
+ /**
+ * Get the value from config by the path. If the path not exist, return the default value.
+ *
+ * @param config The com.vesoft.exchange.common.config.
+ * @param path The path of the com.vesoft.exchange.common.config.
+ * @param defaultValue The default value for the path.
+ * @return
+ */
+ private[this] def getStringOrElse(config: Config, path: String, defaultValue: String): String = {
+ if (config.hasPath(path)) {
+ config.getString(path)
+ } else {
+ defaultValue
+ }
+ }
+
  /**
  * Get the String value from config
  *

exchange-common/src/main/scala/com/vesoft/exchange/common/utils/NebulaUtils.scala

@@ -74,22 +74,22 @@ object NebulaUtils {
  s = s.replaceAll("\\\\", "\\\\\\\\")
  }
  if (s.contains("\t")) {
- s = s.replaceAll("\t", "\\\t")
+ s = s.replaceAll("\t", "\\\\t")
  }
  if (s.contains("\n")) {
- s = s.replaceAll("\n", "\\\n")
+ s = s.replaceAll("\n", "\\\\n")
  }
  if (s.contains("\"")) {
- s = s.replaceAll("\"", "\\\"")
+ s = s.replaceAll("\"", "\\\\\"")
  }
  if (s.contains("\'")) {
- s = s.replaceAll("\'", "\\\'")
+ s = s.replaceAll("\'", "\\\\'")
  }
  if (s.contains("\r")) {
- s = s.replaceAll("\r", "\\\r")
+ s = s.replaceAll("\r", "\\\\r")
  }
  if (s.contains("\b")) {
- s = s.replaceAll("\b", "\\\b")
+ s = s.replaceAll("\b", "\\\\b")
  }
  s
  }

nebula-exchange_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala

@@ -439,12 +439,15 @@ object Exchange {
  private[this] def repartition(frame: DataFrame,
  partition: Int,
  sourceCategory: SourceCategory.Value): DataFrame = {
- val currentPart = frame.rdd.partitions.length
- if (partition > 0 && currentPart != partition
- && !CheckPointHandler.checkSupportResume(sourceCategory)) {
- frame.repartition(partition).toDF
- } else {
+ if (frame.isStreaming || partition <= 0 || CheckPointHandler.checkSupportResume(sourceCategory)) {
  frame
+ } else {
+ val currentPart = frame.rdd.partitions.length
+ if (currentPart == partition) {
+ frame
+ } else {
+ frame.repartition(partition).toDF
+ }
  }
  }
 

nebula-exchange_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala

@@ -322,7 +322,8 @@ object Exchange {
  s">>>>>> total client recordsFailure:${totalClientRecordFailure} \n" +
  s">>>>>> total SST failure:${totalSstRecordFailure} \n" +
  s">>>>>> total SST Success:${totalSstRecordSuccess}")
- LOG.info(s">>>>>> exchange import qps: ${(totalClientRecordSuccess/duration).formatted("%.2f")}/s")
+ LOG.info(
+ s">>>>>> exchange import qps: ${(totalClientRecordSuccess / duration).formatted("%.2f")}/s")
  }
 
  /**
@@ -436,12 +437,15 @@ object Exchange {
  private[this] def repartition(frame: DataFrame,
  partition: Int,
  sourceCategory: SourceCategory.Value): DataFrame = {
- val currentPart = frame.rdd.partitions.length
- if (partition > 0 && currentPart != partition
- && !CheckPointHandler.checkSupportResume(sourceCategory)) {
- frame.repartition(partition).toDF
- } else {
+ if (frame.isStreaming || partition <= 0 || CheckPointHandler.checkSupportResume(sourceCategory)) {
  frame
+ } else {
+ val currentPart = frame.rdd.partitions.length
+ if (currentPart == partition) {
+ frame
+ } else {
+ frame.repartition(partition).toDF
+ }
  }
  }
 

nebula-exchange_spark_3.0/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala

@@ -435,12 +435,15 @@ object Exchange {
  private[this] def repartition(frame: DataFrame,
  partition: Int,
  sourceCategory: SourceCategory.Value): DataFrame = {
- val currentPart = frame.rdd.partitions.length
- if (partition > 0 && currentPart != partition
- && !CheckPointHandler.checkSupportResume(sourceCategory)) {
- frame.repartition(partition).toDF
- } else {
+ if (frame.isStreaming || partition <= 0 || CheckPointHandler.checkSupportResume(sourceCategory)) {
  frame
+ } else {
+ val currentPart = frame.rdd.partitions.length
+ if (currentPart == partition) {
+ frame
+ } else {
+ frame.repartition(partition).toDF
+ }
  }
  }