通過 pulsar-flink-connector 讀取到 Apache pulsar 中的namespaces、topics的元數據信息。pulsar-flink-connector 的 github： https://github.com/streamnative/pulsar-flink

<dependency> <groupId>io.streamnative.connectors</groupId> <artifactId>pulsar-flink-connector-2.11-1.12</artifactId> <version>2.7.3</version> </dependency> <!-- JAR repositories --> <repositories><repository> <id>central</id> <layout>default</layout> <url>https://repo1.maven.org/maven2</url></repository><repository> <id>bintray-streamnative-maven</id> <name>bintray</name> <url>https://dl.bintray.com/streamnative/maven</url></repository> </repositories>CODE

使用PulsarMetadataReader獲取元數據

package com.levi.demo;import org.apache.flink.streaming.connectors.pulsar.internal.PulsarMetadataReader;import org.apache.pulsar.client.admin.PulsarAdminException;import org.apache.pulsar.client.impl.auth.AuthenticationToken;import org.apache.pulsar.client.impl.conf.ClientConfigurationData;import org.apache.pulsar.common.schema.SchemaInfo;import org.apache.pulsar.common.schema.SchemaType;import java.io.IOException;import java.util.HashMap;import java.util.List;import java.util.Map;/** * Test. * * @author levi * @version 1.0 **/public class Test { public static void main(String[] args) {final ClientConfigurationData configurationData = new ClientConfigurationData();configurationData.setServiceUrl('pulsar://127.0.0.1:6650');//Your Pulsar Tokenfinal AuthenticationToken token =new AuthenticationToken('eyJxxxxxxxxxxx.eyxxxxxxxxxxxxx.xxxxxxxxxxx'); configurationData.setAuthentication(token); try (final PulsarMetadataReader reader = new PulsarMetadataReader('http://127.0.0.1:8443', configurationData, '', new HashMap(), -1, -1)) { //獲取namespaces final List<String> namespaces = reader.listNamespaces(); System.out.println('namespaces: ' + namespaces.toString());for (final String namespace : namespaces) {//獲取Topicsfinal List<String> topics = reader.getTopics(namespace);System.out.println('topic: ' + topics.toString());for (String topic : topics) { //獲取字段SchemaInfo final SchemaInfo schemaInfo = reader.getPulsarSchema(topic); final String name = schemaInfo.getName(); System.out.println('SchemaName:' + name); //topicName final SchemaType type = schemaInfo.getType(); System.out.println('SchemaType:' + type.toString());// 'JSON'... final Map<String, String> properties = schemaInfo.getProperties(); System.out.println(properties); final String schemaDefinition = schemaInfo.getSchemaDefinition(); System.out.println(schemaDefinition); // Field info.} }} catch (IOException | PulsarAdminException e) { e.printStackTrace();} }}

到此這篇關于Java使用pulsar-flink-connector讀取pulsar catalog元數據的文章就介紹到這了,更多相關Java讀取pulsar catalog元數據內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網！