通過 pulsar-flink-connector 讀取到 Apache pulsar 中的namespaces、topics的元數(shù)據(jù)信息。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獲取元數(shù)據(jù)

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();} }}

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