[Improve][Connector-V2] Flink support embedding transform (#7592)

* [Improve][Connector-V2] Flink support embedding transform * [Improve][Connector-V2] Optimized code
apache · Sep 6, 2024 · f7286b7 · f7286b7
1 parent b4dbccf
commit f7286b7
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 6 deletions.
diff --git a/docs/en/transform-v2/embedding.md b/docs/en/transform-v2/embedding.md
@@ -320,7 +320,7 @@ transform {
             inputx = ["${input}"]
         }
     }
-    result_table_name = "embedding_output_3"
+    result_table_name = "embedding_output_1"
   }
 }
 

diff --git a/docs/zh/transform-v2/embedding.md b/docs/zh/transform-v2/embedding.md
@@ -310,7 +310,7 @@ transform {
             inputx = ["${input}"]
         }
     }
-    result_table_name = "embedding_output_3"
+    result_table_name = "embedding_output_1"
   }
 }
 

diff --git a/...forms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestEmbeddingIT.java b/...forms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestEmbeddingIT.java
@@ -43,8 +43,8 @@
 
 @DisabledOnContainer(
         value = {},
-        type = {EngineType.SPARK, EngineType.FLINK},
-        disabledReason = "Currently SPARK and FLINK not support adapt")
+        type = {EngineType.SPARK},
+        disabledReason = "Currently SPARK not support adapt")
 public class TestEmbeddingIT extends TestSuiteBase implements TestResource {
     private static final String TMP_DIR = "/tmp";
     private GenericContainer<?> mockserverContainer;

diff --git a/...e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/embedding_transform_custom.conf b/...e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/embedding_transform_custom.conf
@@ -79,10 +79,10 @@ transform {
         }
         custom_request_body ={
             modelx = "${model}"
-            inputx = ["${input}","${input}"]
+            inputx = ["${input}"]
         }
     }
-    result_table_name = "embedding_output_3"
+    result_table_name = "embedding_output_1"
   }
 }
 

diff --git a/...2/src/main/java/org/apache/seatunnel/transform/nlpmodel/embadding/EmbeddingTransform.java b/...2/src/main/java/org/apache/seatunnel/transform/nlpmodel/embadding/EmbeddingTransform.java
@@ -61,6 +61,12 @@ public EmbeddingTransform(
                 config.get(EmbeddingTransformConfig.VECTORIZATION_FIELDS));
     }
 
+    private void tryOpen() {
+        if (model == null) {
+            open();
+        }
+    }
+
     @Override
     public void open() {
         // Initialize model
@@ -161,6 +167,7 @@ private void initOutputFields(SeaTunnelRowType inputRowType, Map<String, String>
 
     @Override
     protected Object[] getOutputFieldValues(SeaTunnelRowAccessor inputRow) {
+        tryOpen();
         try {
             Object[] fieldArray = new Object[fieldOriginalIndexes.size()];
             for (int i = 0; i < fieldOriginalIndexes.size(); i++) {
-Original file line number
+Diff line change
@@ Expand Up / @@ -320,7 +320,7 @@ transform { @@
                 inputx = ["${input}"]
             }
         }
-        result_table_name = "embedding_output_3"
+        result_table_name = "embedding_output_1"
       }
     }
@@ Expand Down @@