pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] sijie closed pull request #3268: Optimization of binlog to pulsar use canal
Date Wed, 02 Jan 2019 09:14:26 GMT
sijie closed pull request #3268: Optimization of binlog to pulsar use canal
URL: https://github.com/apache/pulsar/pull/3268
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalAbstractSource.java
similarity index 81%
rename from pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java
rename to pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalAbstractSource.java
index c348964ded..c1bcb3d504 100644
--- a/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java
+++ b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalAbstractSource.java
@@ -18,20 +18,16 @@
  */
 package org.apache.pulsar.io.canal;
 
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.serializer.SerializerFeature;
 import com.alibaba.otter.canal.client.CanalConnector;
 import com.alibaba.otter.canal.client.CanalConnectors;
-import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.FlatMessage;
+import com.alibaba.otter.canal.protocol.Message;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.PushSource;
 import org.apache.pulsar.io.core.SourceContext;
-import org.apache.pulsar.io.core.annotations.Connector;
-import org.apache.pulsar.io.core.annotations.IOType;
 import org.slf4j.MDC;
 
 import java.net.InetSocketAddress;
@@ -40,16 +36,12 @@
 import java.util.Objects;
 import java.util.Optional;
 
+
 /**
- * A Simple class for mysql binlog sync to pulsar.
+ * A Simple abstract class for mysql binlog sync to pulsar.
  */
-@Connector(
-    name = "canal",
-    type = IOType.SOURCE,
-    help = "The CanalSource is used for syncing mysql binlog to Pulsar.",
-    configClass = CanalSourceConfig.class)
 @Slf4j
-public class CanalSource extends PushSource<byte[]> {
+public abstract class CanalAbstractSource<V> extends PushSource<V> {
 
     protected Thread thread = null;
 
@@ -131,10 +123,9 @@ protected void process() {
                 connector.subscribe();
                 while (running) {
                     Message message = connector.getWithoutAck(canalSourceConfig.getBatchSize());
-                    // delete the setRaw in new version of canal-client
                     message.setRaw(false);
-                    List<FlatMessage> flatMessages = FlatMessage.messageConverter(message);
-                    long batchId = message.getId();
+                    List<FlatMessage> flatMessages = MessageUtils.messageConverter(message);
+                    long batchId = getMessageId(message);
                     int size = message.getEntries().size();
                     if (batchId == -1 || size == 0) {
                         try {
@@ -143,10 +134,9 @@ protected void process() {
                         }
                     } else {
                         if (flatMessages != null) {
-                            CanalRecord canalRecord = new CanalRecord(connector);
-                            String m = JSON.toJSONString(flatMessages, SerializerFeature.WriteMapNullValue);
+                            CanalRecord<V> canalRecord = new CanalRecord<>(connector);
                             canalRecord.setId(batchId);
-                            canalRecord.setRecord(m.getBytes());
+                            canalRecord.setRecord(extractValue(flatMessages));
                             consume(canalRecord);
                         }
                     }
@@ -160,11 +150,15 @@ protected void process() {
         }
     }
 
+    public abstract Long getMessageId(Message message);
+
+    public abstract V extractValue(List<FlatMessage> flatMessages);
+
     @Getter
     @Setter
-    static private class CanalRecord implements Record<byte[]> {
+    static private class CanalRecord<V> implements Record<V> {
 
-        private byte[] record;
+        private V record;
         private Long id;
         private CanalConnector connector;
 
@@ -173,17 +167,19 @@ public CanalRecord(CanalConnector connector) {
         }
 
         @Override
-        public Optional<String>  getKey() {
-            return  Optional.of(Long.toString(id));
+        public Optional<String> getKey() {
+            return Optional.of(Long.toString(id));
         }
 
         @Override
-        public byte[] getValue() {
+        public V getValue() {
             return record;
         }
 
         @Override
-        public Optional<Long> getRecordSequence() {return Optional.of(id);}
+        public Optional<Long> getRecordSequence() {
+            return Optional.of(id);
+        }
 
         @Override
         public void ack() {
@@ -192,5 +188,4 @@ public void ack() {
         }
 
     }
-
-}
+}
\ No newline at end of file
diff --git a/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalByteSource.java
b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalByteSource.java
new file mode 100644
index 0000000000..9e14715724
--- /dev/null
+++ b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalByteSource.java
@@ -0,0 +1,51 @@
+/**
+ * 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.pulsar.io.canal;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.alibaba.otter.canal.protocol.Message;
+import com.alibaba.otter.canal.protocol.FlatMessage;
+import org.apache.pulsar.io.core.annotations.Connector;
+import org.apache.pulsar.io.core.annotations.IOType;
+
+import java.util.List;
+
+/**
+ * A Simple class for mysql binlog sync to pulsar.
+ */
+@Connector(
+    name = "canal",
+    type = IOType.SOURCE,
+    help = "The CanalByteSource is used for syncing mysql binlog to Pulsar.",
+    configClass = CanalSourceConfig.class)
+public class CanalByteSource extends CanalAbstractSource<byte[]> {
+
+    @Override
+    public Long getMessageId(Message message) {
+        return message.getId();
+    }
+
+    @Override
+    public byte[] extractValue(List<FlatMessage> flatMessages) {
+        String messages = JSON.toJSONString(flatMessages, SerializerFeature.WriteMapNullValue);
+        return messages.getBytes();
+    }
+
+}
diff --git a/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalStringSource.java
b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalStringSource.java
new file mode 100644
index 0000000000..2f9bb5fc1e
--- /dev/null
+++ b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalStringSource.java
@@ -0,0 +1,75 @@
+/**
+ * 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.pulsar.io.canal;
+
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.alibaba.otter.canal.protocol.FlatMessage;
+import com.alibaba.otter.canal.protocol.Message;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import com.alibaba.fastjson.JSON;
+import org.apache.pulsar.io.core.annotations.Connector;
+import org.apache.pulsar.io.core.annotations.IOType;
+
+/**
+ * A Simple class for mysql binlog sync to pulsar.
+ */
+@Connector(
+        name = "canal",
+        type = IOType.SOURCE,
+        help = "The CanalStringSource is used for syncing mysql binlog to Pulsar, easy to
use presto sql search.",
+        configClass = CanalSourceConfig.class)
+public class CanalStringSource extends CanalAbstractSource<CanalMessage> {
+
+    private Long messageId;
+
+    @Override
+    public Long getMessageId(Message message) {
+        this.messageId = message.getId();
+        return this.messageId;
+    }
+
+    @Override
+    public CanalMessage extractValue(List<FlatMessage> flatMessages) {
+        String messages = JSON.toJSONString(flatMessages, SerializerFeature.WriteMapNullValue);
+        CanalMessage canalMessage = new CanalMessage();
+        Date date = new Date();
+        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        canalMessage.setTimestamp(dateFormat.format(date));
+        canalMessage.setId(this.messageId);
+        canalMessage.setMessage(messages);
+        return canalMessage;
+    }
+
+}
+
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+class CanalMessage {
+    private Long id;
+    private String message;
+    private String timestamp;
+}
diff --git a/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/MessageUtils.java b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/MessageUtils.java
new file mode 100644
index 0000000000..10586ce5c9
--- /dev/null
+++ b/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/MessageUtils.java
@@ -0,0 +1,161 @@
+/**
+ * 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.pulsar.io.canal;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+
+import com.alibaba.otter.canal.protocol.CanalEntry;
+import com.alibaba.otter.canal.protocol.FlatMessage;
+import com.alibaba.otter.canal.protocol.Message;
+import com.google.protobuf.ByteString;
+
+/**
+ * A Simple class for mysql binlog message to parse.
+ */
+public class MessageUtils {
+
+    public static Map<String, String> genColumn(CanalEntry.Column column) {
+        Map<String, String> row = new LinkedHashMap<>();
+        if (column.getIsKey()) {
+            row.put("isKey", "1");
+        } else {
+            row.put("isKey", "0");
+        }
+        if (column.getIsNull()) {
+            row.put("isNull", "1");
+        } else {
+            row.put("isNull", "0");
+        }
+        row.put("index", Integer.toString(column.getIndex()));
+        row.put("mysqlType", column.getMysqlType());
+        row.put("columnName", column.getName());
+        if (column.getIsNull()) {
+            row.put("columnValue", null);
+        } else {
+            row.put("columnValue", column.getValue());
+        }
+        return row;
+    }
+
+    /**
+     * Message convert to FlatMessage
+     *
+     * @param message
+     * @return FlatMessage List
+     */
+    public static List<FlatMessage> messageConverter(Message message) {
+        try {
+            if (message == null) {
+                return null;
+            }
+
+            List<FlatMessage> flatMessages = new ArrayList<>();
+            List<CanalEntry.Entry> entrys = null;
+            if (message.isRaw()) {
+                List<ByteString> rawEntries = message.getRawEntries();
+                entrys = new ArrayList<CanalEntry.Entry>(rawEntries.size());
+                for (ByteString byteString : rawEntries) {
+                    CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString);
+                    entrys.add(entry);
+                }
+            } else {
+                entrys = message.getEntries();
+            }
+
+            for (CanalEntry.Entry entry : entrys) {
+                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
+                        || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
+                    continue;
+                }
+
+                CanalEntry.RowChange rowChange;
+                try {
+                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
+                } catch (Exception e) {
+                    throw new RuntimeException("ERROR ## parser of eromanga-event has an
error , data:"
+                            + entry.toString(), e);
+                }
+
+                CanalEntry.EventType eventType = rowChange.getEventType();
+
+                FlatMessage flatMessage = new FlatMessage(message.getId());
+                flatMessages.add(flatMessage);
+                flatMessage.setDatabase(entry.getHeader().getSchemaName());
+                flatMessage.setTable(entry.getHeader().getTableName());
+                flatMessage.setIsDdl(rowChange.getIsDdl());
+                flatMessage.setType(eventType.toString());
+                flatMessage.setEs(entry.getHeader().getExecuteTime());
+                flatMessage.setTs(System.currentTimeMillis());
+                flatMessage.setSql(rowChange.getSql());
+
+                if (!rowChange.getIsDdl()) {
+                    List<Map<String, String>> data = new ArrayList<>();
+                    List<Map<String, String>> old = new ArrayList<>();
+
+                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
+                        if (eventType != CanalEntry.EventType.INSERT && eventType
!= CanalEntry.EventType.UPDATE
+                                && eventType != CanalEntry.EventType.DELETE) {
+                            continue;
+                        }
+
+                        List<CanalEntry.Column> columns;
+
+                        if (eventType == CanalEntry.EventType.DELETE) {
+                            columns = rowData.getBeforeColumnsList();
+                        } else {
+                            columns = rowData.getAfterColumnsList();
+                        }
+                        columns.size();
+                        for (CanalEntry.Column column : columns) {
+                            Map<String, String> row = genColumn(column);
+                            if (column.getUpdated()) {
+                                row.put("updated", "1");
+                            } else {
+                                row.put("updated", "0");
+                            }
+                            data.add(row);
+                        }
+
+                        if (eventType == CanalEntry.EventType.UPDATE) {
+                            for (CanalEntry.Column column : rowData.getBeforeColumnsList())
{
+                                Map<String, String> rowOld = genColumn(column);
+                                old.add(rowOld);
+                            }
+                        }
+                    }
+
+                    if (!data.isEmpty()) {
+                        flatMessage.setData(data);
+                    }
+                    if (!old.isEmpty()) {
+                        flatMessage.setOld(old);
+                    }
+                }
+            }
+            return flatMessages;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+}
diff --git a/pulsar-io/canal/src/main/resources/META-INF/services/pulsar-io.yaml b/pulsar-io/canal/src/main/resources/META-INF/services/pulsar-io.yaml
index fa48f3906f..2ecf4df594 100644
--- a/pulsar-io/canal/src/main/resources/META-INF/services/pulsar-io.yaml
+++ b/pulsar-io/canal/src/main/resources/META-INF/services/pulsar-io.yaml
@@ -19,4 +19,4 @@
 
 name: canal
 description: canal source and read data from mysql
-sourceClass: org.apache.pulsar.io.canal.CanalSource
+sourceClass: org.apache.pulsar.io.canal.CanalStringSource


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message