add error amount limit (#58)

* duration for sst * add error amount limit
vesoft-inc · Jan 17, 2022 · 4741be3 · 4741be3
1 parent 2785af0
commit 4741be3
Show file tree

Hide file tree

Showing 7 changed files with 39 additions and 0 deletions.
diff --git a/exchange-common/src/main/scala/com/vesoft/exchange/common/processor/Processor.scala b/exchange-common/src/main/scala/com/vesoft/exchange/common/processor/Processor.scala
@@ -179,6 +179,9 @@ trait Processor extends Serializable {
         val jtsGeom = new org.locationtech.jts.io.WKTReader().read(wkt)
         convertJTSGeometryToGeography(jtsGeom)
       }
+      case PropertyType.DURATION => {
+        throw new IllegalArgumentException("do not support data type duration.")
+      }
     }
   }
 

diff --git a/...xchange_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala b/...xchange_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala
@@ -22,6 +22,7 @@ import com.vesoft.exchange.common.utils.NebulaUtils.DEFAULT_EMPTY_VALUE
 import com.vesoft.exchange.common.writer.{GenerateSstFile, NebulaGraphClientWriter, NebulaSSTWriter}
 import com.vesoft.exchange.common.VidType
 import com.vesoft.nebula.encoder.NebulaCodecImpl
+import com.vesoft.nebula.exchange.TooManyErrorsException
 import com.vesoft.nebula.meta.EdgeItem
 import org.apache.commons.codec.digest.MurmurHash2
 import org.apache.log4j.Logger
@@ -72,6 +73,11 @@ class EdgeProcessor(spark: SparkSession,
       } else {
         errorBuffer.append(failStatement)
         batchFailure.add(1)
+        if (batchFailure.value >= config.errorConfig.errorMaxSize) {
+          throw TooManyErrorsException(
+            s"There are too many failed batches, batch amount: ${batchFailure.value}, " +
+              s"your config max error size: ${config.errorConfig.errorMaxSize}")
+        }
       }
     }
     if (errorBuffer.nonEmpty) {

diff --git a/...nge_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/processor/VerticesProcessor.scala b/...nge_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/processor/VerticesProcessor.scala
@@ -21,6 +21,7 @@ import com.vesoft.exchange.common.utils.NebulaUtils.DEFAULT_EMPTY_VALUE
 import com.vesoft.exchange.common.writer.{GenerateSstFile, NebulaGraphClientWriter, NebulaSSTWriter}
 import com.vesoft.exchange.common.VidType
 import com.vesoft.nebula.encoder.NebulaCodecImpl
+import com.vesoft.nebula.exchange.TooManyErrorsException
 import com.vesoft.nebula.meta.TagItem
 import org.apache.commons.codec.digest.MurmurHash2
 import org.apache.log4j.Logger
@@ -79,6 +80,11 @@ class VerticesProcessor(spark: SparkSession,
       } else {
         errorBuffer.append(failStatement)
         batchFailure.add(1)
+        if (batchFailure.value >= config.errorConfig.errorMaxSize) {
+          throw TooManyErrorsException(
+            s"There are too many failed batches, batch amount: ${batchFailure.value}, " +
+              s"your config max error size: ${config.errorConfig.errorMaxSize}")
+        }
       }
     }
     if (errorBuffer.nonEmpty) {

diff --git a/...xchange_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala b/...xchange_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala
@@ -23,6 +23,7 @@ import com.vesoft.exchange.common.utils.NebulaUtils.DEFAULT_EMPTY_VALUE
 import com.vesoft.exchange.common.writer.{GenerateSstFile, NebulaGraphClientWriter, NebulaSSTWriter}
 import com.vesoft.exchange.common.VidType
 import com.vesoft.nebula.encoder.NebulaCodecImpl
+import com.vesoft.nebula.exchange.TooManyErrorsException
 import com.vesoft.nebula.meta.EdgeItem
 import org.apache.commons.codec.digest.MurmurHash2
 import org.apache.log4j.Logger
@@ -73,6 +74,11 @@ class EdgeProcessor(spark: SparkSession,
       } else {
         errorBuffer.append(failStatement)
         batchFailure.add(1)
+        if (batchFailure.value >= config.errorConfig.errorMaxSize) {
+          throw TooManyErrorsException(
+            s"There are too many failed batches, batch amount: ${batchFailure.value}, " +
+              s"your config max error size: ${config.errorConfig.errorMaxSize}")
+        }
       }
     }
     if (errorBuffer.nonEmpty) {

diff --git a/...nge_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/processor/VerticesProcessor.scala b/...nge_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/processor/VerticesProcessor.scala
@@ -22,6 +22,7 @@ import com.vesoft.exchange.common.utils.NebulaUtils.DEFAULT_EMPTY_VALUE
 import com.vesoft.exchange.common.writer.{GenerateSstFile, NebulaGraphClientWriter, NebulaSSTWriter}
 import com.vesoft.exchange.common.VidType
 import com.vesoft.nebula.encoder.NebulaCodecImpl
+import com.vesoft.nebula.exchange.TooManyErrorsException
 import com.vesoft.nebula.meta.TagItem
 import org.apache.commons.codec.digest.MurmurHash2
 import org.apache.log4j.Logger
@@ -81,6 +82,11 @@ class VerticesProcessor(spark: SparkSession,
       } else {
         errorBuffer.append(failStatement)
         batchFailure.add(1)
+        if (batchFailure.value >= config.errorConfig.errorMaxSize) {
+          throw TooManyErrorsException(
+            s"There are too many failed batches, batch amount: ${batchFailure.value}, " +
+              s"your config max error size: ${config.errorConfig.errorMaxSize}")
+        }
       }
     }
     if (errorBuffer.nonEmpty) {

diff --git a/...xchange_spark_3.0/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala b/...xchange_spark_3.0/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala
@@ -23,6 +23,7 @@ import com.vesoft.exchange.common.utils.NebulaUtils.DEFAULT_EMPTY_VALUE
 import com.vesoft.exchange.common.writer.{GenerateSstFile, NebulaGraphClientWriter, NebulaSSTWriter}
 import com.vesoft.exchange.common.VidType
 import com.vesoft.nebula.encoder.NebulaCodecImpl
+import com.vesoft.nebula.exchange.TooManyErrorsException
 import com.vesoft.nebula.meta.EdgeItem
 import org.apache.commons.codec.digest.MurmurHash2
 import org.apache.log4j.Logger
@@ -73,6 +74,11 @@ class EdgeProcessor(spark: SparkSession,
       } else {
         errorBuffer.append(failStatement)
         batchFailure.add(1)
+        if (batchFailure.value >= config.errorConfig.errorMaxSize) {
+          throw TooManyErrorsException(
+            s"There are too many failed batches, batch amount: ${batchFailure.value}, " +
+              s"your config max error size: ${config.errorConfig.errorMaxSize}")
+        }
       }
     }
     if (errorBuffer.nonEmpty) {

diff --git a/...nge_spark_3.0/src/main/scala/com/vesoft/nebula/exchange/processor/VerticesProcessor.scala b/...nge_spark_3.0/src/main/scala/com/vesoft/nebula/exchange/processor/VerticesProcessor.scala
@@ -22,6 +22,7 @@ import com.vesoft.exchange.common.utils.NebulaUtils.DEFAULT_EMPTY_VALUE
 import com.vesoft.exchange.common.writer.{GenerateSstFile, NebulaGraphClientWriter, NebulaSSTWriter}
 import com.vesoft.exchange.common.VidType
 import com.vesoft.nebula.encoder.NebulaCodecImpl
+import com.vesoft.nebula.exchange.TooManyErrorsException
 import com.vesoft.nebula.meta.TagItem
 import org.apache.commons.codec.digest.MurmurHash2
 import org.apache.log4j.Logger
@@ -81,6 +82,11 @@ class VerticesProcessor(spark: SparkSession,
       } else {
         errorBuffer.append(failStatement)
         batchFailure.add(1)
+        if (batchFailure.value >= config.errorConfig.errorMaxSize) {
+          throw TooManyErrorsException(
+            s"There are too many failed batches, batch amount: ${batchFailure.value}, " +
+              s"your config max error size: ${config.errorConfig.errorMaxSize}")
+        }
       }
     }
     if (errorBuffer.nonEmpty) {