Java使用pulsar-flink-connector读取pulsar catalog元数据代码剖析
简介
通过 pulsar-flink-connector 读取到 Apache pulsar 中的namespaces、topics的元数据信息。
pulsar-flink-connector 的 github: https://github.com/streamnative/pulsar-flink
Maven
<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 Token final 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) { //获取Topics final 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元数据内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!
相关文章
关于Springboot的扩展点DisposableBean的原理解析
这篇文章主要介绍了关于Springboot的扩展点DisposableBean的原理解析,DisposableBean是一个接口,为Spring bean提供了一种释放资源的方式 ,只有一个扩展方法destroy(),需要的朋友可以参考下2023-05-05Java SpringMVC 异常处理SimpleMappingExceptionResolver类详解
这篇文章主要介绍了SpringMVC 异常处理SimpleMappingExceptionResolver类详解,本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下2021-09-09java模拟http的Get/Post请求,并设置ip与port代理的方法
下面小编就为大家带来一篇java模拟http的Get/Post请求,并设置ip与port代理的方法。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧2017-02-02
最新评论