EricJoy2048 commented on code in PR #3673:
URL:
https://github.com/apache/incubator-seatunnel/pull/3673#discussion_r1042232972
##########
seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java:
##########
@@ -66,25 +116,20 @@ public String serializeRow(SeaTunnelRow row) {
doc.put(fieldNames[i], value);
}
}
+ return doc;
+ }
- StringBuilder sb = new StringBuilder();
-
- Map<String, String> indexInner = new HashMap<>();
- String index = indexSerializer.serialize(row);
- indexInner.put("_index", index);
- indexTypeSerializer.fillType(indexInner);
-
- Map<String, Map<String, String>> indexParam = new HashMap<>();
- indexParam.put("index", indexInner);
- try {
- sb.append(objectMapper.writeValueAsString(indexParam));
- sb.append("\n");
- String indexDoc = objectMapper.writeValueAsString(doc);
- sb.append(indexDoc);
- } catch (JsonProcessingException e) {
- throw new
ElasticsearchConnectorException(CommonErrorCode.JSON_OPERATION_FAILED, "Object
json deserialization exception.", e);
- }
+ private Map<String, String> createMetadata(@NonNull SeaTunnelRow row,
+ @NonNull String key) {
+ Map<String, String> actionMetadata = createMetadata(row);
+ actionMetadata.put("_id", key);
+ return actionMetadata;
+ }
- return sb.toString();
+ private Map<String, String> createMetadata(@NonNull SeaTunnelRow row) {
+ Map<String, String> actionMetadata = new HashMap<>();
Review Comment:
gave a init capacity？
##########
seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java:
##########
@@ -45,15 +47,63 @@ public class ElasticsearchRowSerializer implements
SeaTunnelRowSerializer {
private final IndexSerializer indexSerializer;
private final IndexTypeSerializer indexTypeSerializer;
+ private final Function<SeaTunnelRow, String> keyExtractor;
public ElasticsearchRowSerializer(ElasticsearchVersion
elasticsearchVersion, IndexInfo indexInfo, SeaTunnelRowType seaTunnelRowType) {
this.indexTypeSerializer =
IndexTypeSerializerFactory.getIndexTypeSerializer(elasticsearchVersion,
indexInfo.getType());
this.indexSerializer =
IndexSerializerFactory.getIndexSerializer(indexInfo.getIndex(),
seaTunnelRowType);
this.seaTunnelRowType = seaTunnelRowType;
+ this.keyExtractor = KeyExtractor.createKeyExtractor(seaTunnelRowType,
indexInfo.getPrimaryKeys(), indexInfo.getKeyDelimiter());
}
@Override
public String serializeRow(SeaTunnelRow row) {
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ return serializeUpsert(row);
+ case UPDATE_BEFORE:
+ case DELETE:
+ return serializeDelete(row);
+ default:
+ throw new ElasticsearchConnectorException(
+ CommonErrorCode.UNSUPPORTED_OPERATION, "Unsupported write
row kind: " + row.getRowKind());
+ }
+ }
+
+ private String serializeUpsert(SeaTunnelRow row) {
+ String key = keyExtractor.apply(row);
+ Map<String, Object> document = toDocumentMap(row);
+
+ try {
+ if (key != null) {
+ Map<String, String> upsertMetadata = createMetadata(row, key);
+ return String.format("{ \"update\" : %s }\n{ \"doc\" : %s,
\"doc_as_upsert\" : true }",
Review Comment:
`serializeUpsert ` will be called by every row. So, I suggest we use better
way to process String.
##########
seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractor.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.elasticsearch.serialize;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.exception.CommonErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorException;
+
+import lombok.AllArgsConstructor;
+
+import java.io.Serializable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+
+@AllArgsConstructor
+public class KeyExtractor implements Function<SeaTunnelRow, String>,
Serializable {
+ private final FieldFormatter[] fieldFormatters;
+ private final String keyDelimiter;
+
+ @Override
+ public String apply(SeaTunnelRow row) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < fieldFormatters.length; i++) {
+ if (i > 0) {
+ builder.append(keyDelimiter);
+ }
+ String value = fieldFormatters[i].format(row);
+ builder.append(value);
+ }
+ return builder.toString();
+ }
+
+ public static Function<SeaTunnelRow, String>
createKeyExtractor(SeaTunnelRowType rowType,
+ String[]
primaryKeys,
+ String
keyDelimiter) {
+ if (primaryKeys == null) {
+ return row -> null;
+ }
+
+ List<FieldFormatter> fieldFormatters = new ArrayList<>();
Review Comment:
gave an init capacity?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
