[Feature][connector-v2]Add Kafka Protobuf Data Parsing Support #7361

zhangshenghang · 2024-08-10T02:09:41Z

Purpose of this pull request

handle #7059

How was this patch tested?

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update change log that in connector document. For more details you can refer to connector-v2
2. Update plugin-mapping.properties and add new connector information in it
3. Update the pom file of seatunnel-dist
Update the release-note.

zhangshenghang · 2024-08-10T02:37:06Z

@Hisoka-X @hailin0
By configuring the protobuf data structure, we can successfully receive and send related messages.

This is an example used, and I will continue to improve the relevant documentation and unit testing.

Help me see where we need to optimize , thk.

# Defining the runtime environment
env {
    parallelism = 1
    job.mode = "STREAMING"
    flink.execution.checkpointing.interval=5000
     flink.execution.restart.strategy = failure-rate
     flink.execution.restart.failureInterval = 60000
     flink.execution.restart.failureRate = 100
     flink.execution.restart.delayInterval = 10000
   # execution.restart.strategy = fixed-delay
   # execution.restart.attempts = 11
   # execution.restart.delayBetweenAttempts = 10000
}
source {
  Kafka {
    schema = {
      fields {
        name = string
        id = int
        email = string
        Address {
            city = string
            state = string
            street = string
            }
        attributes = "map<string,float>"
        }
    }
    topic = "test"
    format = protobuf
    protobuf_message_name = Person
    protobuf_schema = """
              syntax = "proto3";

              package wiki.hadoop.protobuf;

              option java_outer_classname = "Test2";

              message Person {
                string name = 1;
                int32 id = 2;
                string email = 3;

                message Address {
                  string street = 1;
                  string city = 2;
                  string state = 3;
                  string zip = 4;
                }

                Address address = 4;

                map<string, float> attributes = 5;

                repeated string phone_numbers = 6;
              }
    """
    consumer.group	= "test-202301011"
    bootstrap.servers = "172.16.24.194:9092"
    kafka.config = {
      client.id = client_1
      max.poll.records = 500
      auto.offset.reset = "latest"
      enable.auto.commit = "false"
    }
    result_table_name = "kafka_table"
  }
}
transform {
  Sql {
    source_table_name = "kafka_table"
    result_table_name = "kafka_table2"
    query = "select name, Address.state ,Address.city ,attributes.test1 from kafka_table"
  }
}


sink {

    Console{
    source_table_name = "kafka_table2"
    }

}

# Defining the runtime environment
env {
    parallelism = 1
    job.mode = "STREAMING"
    flink.execution.checkpointing.interval=5000
     flink.execution.restart.strategy = failure-rate
     flink.execution.restart.failureInterval = 60000
     flink.execution.restart.failureRate = 100
     flink.execution.restart.delayInterval = 10000

}
source {
   FakeSource {
      parallelism = 1
      result_table_name = "fake"
      row.num = 16
      schema = {
        fields {
                  name = string
                  id = int
                  email = string
                  Address {
                      city = string
                      state = string
                      street = string
                      }
                  attributes = "map<string,float>"
                  phone_numbers = "array<string>"
        }
      }
    }
}

sink {
  Console{}
  kafka {
      topic = "test_topic"
      bootstrap.servers = "172.16.24.194:9092"
      format = protobuf
      kafka.request.timeout.ms = 60000
#       semantics = EXACTLY_ONCE
      kafka.config = {
        acks = "all"
        request.timeout.ms = 60000
        buffer.memory = 33554432
      }
      protobuf_message_name = Person
      protobuf_schema = """
              syntax = "proto3";

              package wiki.hadoop.protobuf;

              option java_outer_classname = "Test2";

              message Person {
                string name = 1;
                int32 id = 2;
                string email = 3;

                message Address {
                  string street = 1;
                  string city = 2;
                  string state = 3;
                  string zip = 4;
                }

                Address address = 4;

                map<string, float> attributes = 5;

                repeated string phone_numbers = 6;
              }
              """
  }
}

Hisoka-X · 2024-08-10T02:47:02Z

protobuf_schema = """
              syntax = "proto3";

              package wiki.hadoop.protobuf;

              option java_outer_classname = "Test2";

              message Person {
                string name = 1;
                int32 id = 2;
                string email = 3;

                message Address {
                  string street = 1;
                  string city = 2;
                  string state = 3;
                  string zip = 4;
                }

                Address address = 4;

                map<string, float> attributes = 5;

                repeated string phone_numbers = 6;
              }
              """

+1 for this configure way. cc @hailin0

zhangshenghang · 2024-08-22T12:48:11Z

Hi @hailin0 , Is this method OK? If so, I will add the documentation and e2e code

Hisoka-X · 2024-08-23T06:20:56Z

...el-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/CompileDescriptor.java

+    }
+
+    private static File createTempDirectory() throws IOException {
+        File tmpDir = File.createTempFile("tmp_protobuf_", "_proto");


Put all protobuf file into /tmp/seatunnel/tmp_protobuf_?

Put all protobuf file into /tmp/seatunnel/tmp_protobuf_?

Protobuf needs to convert configuration content into descriptor files using Protoc.runProtoc conversion, which requires temporary directories and files. When executing the code for the first time, protobuf configuration content will be parsed into descriptor files, which will be used to parse data

I think this does not conflict when placed in the specified temporary path?

I think this does not conflict when placed in the specified temporary path?

File.createTempFile("tmp_protobuf_", "_proto").getPath()
This method generates directories or files completely randomly, and its structure is as follows:
/var/folders/ys/223gp9td7dqf4nmxl9kw25lc0000gn/T/tmp_protobuf_2079053607848123529_proto

ok. I got it. Thanks for explain.

Hisoka-X · 2024-08-23T06:24:03Z

...el-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/CompileDescriptor.java

+        File tmpDir = createTempDirectory();
+        File protoFile = createProtoFile(tmpDir, protoContent);
+        String targetDescPath = compileProtoToDescriptor(tmpDir, protoFile);
+
+        try (FileInputStream fis = new FileInputStream(targetDescPath)) {
+            DescriptorProtos.FileDescriptorSet descriptorSet =
+                    DescriptorProtos.FileDescriptorSet.parseFrom(fis);
+            Descriptors.FileDescriptor[] descriptorsArray = buildFileDescriptors(descriptorSet);
+            return descriptorsArray[0].findMessageTypeByName(messageName);
+        }


can we delete the tmp file in finally block?

can we delete the tmp file in finally block?

tmpDir.deleteOnExit();

It will be automatically deleted when the program exits, without the need to call the deletion method.

But in cluster mode, the program may never exit.

But in cluster mode, the program may never exit.

But in cluster mode, the program may never exit.

You're right, I will fix it. It should indeed be deleted immediately after use

...el-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/CompileDescriptor.java

Hisoka-X

Please add e2e test case with read/writer PROTOBUF format.

Hisoka-X · 2024-08-23T07:02:19Z

...el-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/CompileDescriptor.java

+    }
+
+    private static File createTempDirectory() throws IOException {
+        File tmpDir = File.createTempFile("tmp_protobuf_", "_proto");


ok. I got it. Thanks for explain.

zhangshenghang · 2024-08-24T15:52:53Z

Please add e2e test case with read/writer PROTOBUF format.

@Hisoka-X Add Complete

Hisoka-X · 2024-08-27T04:14:08Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

+                new HashMap<String, String>() {
+                    {
+                        put(PROTOBUF_MESSAGE_NAME.key(), readonlyConfig.get(PROTOBUF_MESSAGE_NAME));
+                        put(PROTOBUF_SCHEMA.key(), readonlyConfig.get(PROTOBUF_SCHEMA));
+                    }
+                },


If user does not use protobuf, the value would be null? So what's use for?

Hisoka-X · 2024-08-27T04:25:16Z

...nnector-v2-e2e/connector-kafka-e2e/src/test/resources/protobuf/kafka_protobuf_to_assert.conf

+    Console{
+        source_table_name = "kafka_table"
+    }


Please use assert to check data.

Hisoka-X

LGTM. Thanks @zhangshenghang !

wuchunfu

LGTM, thanks

…e#7361) * [feature]add protobuf format * [featrue]fix style * [featrue]fix style * [feature]add protobuf * [feature]add protobuf * [feature]add protobuf * [feature]add protobuf * [feature]add license * [feature]fix some problem * [feature]fix some problem * [feature]fix some problem * [feature]fix some problem * [feature]fix some problem * [feature]merge * [improve]delete useless dependencies from known-dependencies.txt * [feature]add unit test * [feature]add e2e and uint test * [feature]add license and doc * [feature]fix some problem --------- Co-authored-by: hailin0 <[email protected]>

chessplay · 2024-09-02T10:03:21Z

I am working on a streaming job using this feature, where I need to parse a Kafka topic containing a protobuf message with nested objects in an array. Below is the relevant portion of my configuration:

hocon
复制代码
source {
Kafka {
schema = {
fields {
sn = string
tenantId = int
groupId = int
utcCode = int
updateTime = bigint
totalUsers = int
array{CurrentUserInfo {
userId = int
mac = string
userIp = string
rssi = int
uplinkRate = float
upRate = bigint
timeDelay = int
score = int
}
}
}
}
topic = "pb-wirelessUser"
format = protobuf
protobuf_message_name = CurrentUserContent
protobuf_schema = """
syntax = "proto3";
package com.ruijie.cloud.macc.log.proto.sta;

  message CurrentUserContent {
    string sn = 1;
    uint32 tenantId = 2;
    uint32 groupId = 4;
    int32 utcCode = 8;
    int64 updateTime = 9;
    uint32 totalUsers = 11;
    repeated CurrentUserInfo userList = 20;
  }

  message CurrentUserInfo {
    int32 userId = 1;
    string mac = 2;
    string userIp = 3;
    int32 rssi = 7;
    float uplinkRate = 8;
    int64 upRate = 11;
    int32 timeDelay = 17;
    int32 score = 25;
  }
"""
consumer.group = "seatunnel-test"
bootstrap.servers = "192.168.105.102:9092"
result_table_name = "kafka_table"

}
}

transform {
Sql {
source_table_name = "kafka_table"
result_table_name = "kafka_table2"
query = "select CurrentUserInfo from kafka_table"
}
}

sink {
Console {
source_table_name = "kafka_table2"
}
}
The protobuf message has a repeated field CurrentUserInfo as part of the CurrentUserContent message. My question is: How should I configure SeaTunnel to correctly parse and extract the nested CurrentUserInfo array?

I've tried the above configuration, but I'm not sure if this is the correct approach. Any guidance or examples of similar configurations would be greatly appreciated!

Hisoka-X · 2024-09-02T10:07:50Z

@chessplay have you tried https://seatunnel.apache.org/docs/2.3.7/transform-v2/sql#struct-query ?

zhangshenghang · 2024-09-02T12:33:02Z

@chessplay https://github.com/apache/seatunnel/blob/dev/docs/zh/connector-v2/source/Kafka.md
Here are some usage examples

chessplay · 2024-09-04T10:01:52Z

@Hisoka-X I have tried struct-query ，but it does not seems to work。The following is key configuration and log，I would like to ask if this feature does not support handling nested object arrays in JSON?

Hisoka-X · 2024-09-04T10:46:33Z

@chessplay Could you create an issue for this?

chessplay · 2024-09-05T02:25:05Z

ok,i have created an issue for this #7585

zhangshenghang added 13 commits July 15, 2024 14:00

Merge remote-tracking branch 'upstream/dev' into HEAD

dc44978

[feature]add protobuf format

1963ec8

[featrue]fix style

bf14f8f

[featrue]fix style

b43646b

[feature]add protobuf

8685750

[feature]add protobuf

3419e22

[feature]add protobuf

0d5f4af

[feature]add protobuf

064ed36

[feature]add license

ce3ff57

[feature]fix some problem

d5ccf23

[feature]fix some problem

ca89bca

[feature]fix some problem

9431e51

[feature]fix some problem

7e815e6

github-actions bot added dependencies Pull requests that update a dependency file connectors-v2 e2e format kafka labels Aug 10, 2024

zhangshenghang and others added 2 commits August 10, 2024 10:12

[feature]fix some problem

e3526fb

Merge branch 'dev' into feature-add-kafka-data-parsing

2b98dfd

[feature]merge

ac05e8c

Hisoka-X linked an issue Aug 10, 2024 that may be closed by this pull request

[Feature][Kafka-connector] Proposal to Add Kafka Protobuf Data Parsing Support #7059

Closed

3 tasks

zhangshenghang and others added 4 commits August 21, 2024 20:24

Merge branch 'dev' into feature-add-kafka-data-parsing

ee1a094

[improve]delete useless dependencies from known-dependencies.txt

6d05db3

Merge branch 'apache:dev' into feature-add-kafka-data-parsing

371ac4c

Merge branch 'apache:dev' into feature-add-kafka-data-parsing

b1261ce

Hisoka-X reviewed Aug 23, 2024

View reviewed changes

zhangshenghang added 3 commits August 24, 2024 08:02

[feature]add unit test

659ec28

[feature]add e2e and uint test

e82149f

[feature]add license and doc

9db26c6

github-actions bot added the document label Aug 24, 2024

Hisoka-X reviewed Aug 27, 2024

View reviewed changes

[feature]fix some problem

3f45308

Hisoka-X approved these changes Aug 28, 2024

View reviewed changes

github-actions bot added approved reviewed labels Aug 28, 2024

wuchunfu approved these changes Aug 28, 2024

View reviewed changes

wuchunfu merged commit 51c8e1a into apache:dev Aug 28, 2024

zhangshenghang mentioned this pull request Sep 2, 2024

[Docs][kafka] Add kafka protobuf doc #7553

Merged

4 tasks

zhangshenghang deleted the feature-add-kafka-data-parsing branch September 6, 2024 02:32

[Feature][connector-v2]Add Kafka Protobuf Data Parsing Support #7361

[Feature][connector-v2]Add Kafka Protobuf Data Parsing Support #7361

Uh oh!

Conversation

zhangshenghang commented Aug 10, 2024

Purpose of this pull request

How was this patch tested?

Check list

Uh oh!

zhangshenghang commented Aug 10, 2024

Uh oh!

Hisoka-X commented Aug 10, 2024

Uh oh!

zhangshenghang commented Aug 22, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Hisoka-X left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhangshenghang commented Aug 24, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Hisoka-X left a comment

Choose a reason for hiding this comment

Uh oh!

wuchunfu left a comment

Choose a reason for hiding this comment

Uh oh!

chessplay commented Sep 2, 2024

Uh oh!

Hisoka-X commented Sep 2, 2024

Uh oh!

zhangshenghang commented Sep 2, 2024

Uh oh!

chessplay commented Sep 4, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Hisoka-X commented Sep 4, 2024

Uh oh!

chessplay commented Sep 5, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

chessplay commented Sep 4, 2024 •

edited

Loading