首页 > 其他分享 >通过Kafka在ignite集群之间进行实时数据复制

通过Kafka在ignite集群之间进行实时数据复制

时间：2023-06-15 12:01:53浏览次数：55  
标签：ignite Ignite 缓存 Kafka 集群 连接器 kafka


版本1.6的ApacheIgnite提供了一种基于KafkaConnect进行数据处理的新方法。Kafka Connect是ApacheKafka 0.9中引入的一个新特性，它支持ApacheKafka和其他数据系统之间的可伸缩和可靠的流数据。它使得在内存中向您的可伸缩和安全的流数据管道中添加新系统变得非常容易。在本文中，我们将研究如何设置和配置IgniteSource连接器，以便在Ignite集群之间执行数据复制。

apache ignite，开箱即用，提供点燃-卡夫卡模块采用三种不同的解决方案(API)来实现健壮的数据处理管道，将数据从/到Kafka主题流到ApacheIgnite。

通过Kafka在ignite集群之间进行实时数据复制_apache

简而言之，ApacheIgniteSourceConnector用于订阅Ignite缓存事件并将其流到Kafka主题。换句话说，它可以用于从Ignite缓存导出数据(更改的数据集)，并仅使用配置文件将内容写入Kafka主题。点燃源连接器侦听注册的Ignite网格事件(如PUT)，并将它们转发到Kafka主题。这使得保存到Ignite缓存中的数据能够很容易地转换为事件流。每个事件流包含一个键和两个值：旧的和新的。

通过Kafka在ignite集群之间进行实时数据复制_kafka_02

这篇文章的部分内容摘自这本书。阿帕奇·艾格尼特的书。如果它让你感兴趣，看看书的其余部分，以获得更有帮助的信息。

这，这个，那，那个IgniteSourceConnector对于支持以下用例可能很有用：

  1. 若要在发生缓存事件时自动通知任何客户端，例如，每当有新条目进入缓存时。
  2. 若要使用从Ignite缓存流到1-N目的地的异步事件流，请执行以下操作。目标可以是任何数据库，也可以是另一个Ignite集群。这些使您能够通过Kafka在两个Ignite集群之间进行数据复制。

阿帕奇IgniteSourceConnector与IgniteSinkConnector一起使用的工具，可在ignite-kafka-x.x.x.jar分配。IgniteSourceConnector需要以下配置参数：

通过Kafka在ignite集群之间进行实时数据复制_kafka_03

的高级体系结构。IgniteSinkConnector如图1所示。

通过Kafka在ignite集群之间进行实时数据复制_kafka_04

在本文中，我们将使用两个IgniteSourceConnector和IgniteSinkConnector对于从一个Ignite集群到另一个Ignite集群的流事件。IgniteSourceConnector将事件从一个Ignite集群(源集群)流到Kafka主题，IgniteSinkConnector将更改从主题流到另一个Ignite集群(目标集群)。我们将演示逐步配置和运行Source和Sink连接器的说明.为了完成Ignite集群之间的数据复制，我们将执行以下操作：

  1. 在一台机器上执行两个独立的Ignite集群。
  2. 在发送到Ignite目标集群之前，开发一个流提取器来解析传入的数据。
  3. 在不同的独立卡夫卡工人中配置并启动IgniteSource和Sink连接器。
  4. 向Ignite源集群中添加或修改某些数据。

在完成所有配置之后，您应该有一个典型的管道，即将数据从一个Ignite集群流到另一个Ignite集群，如图2所示。

通过Kafka在ignite集群之间进行实时数据复制_xml_05

让我们从Ignite集群配置开始。

第一步。我们将在一台机器上启动两个孤立的集群。要做到这一点，我们必须使用另一组TcpDiscoverySpi和TcpConfigurationSpi将两个集群分离到一个主机上。因此，对于第一个集群中的节点，我们将使用以下方法TcpDiscoverySpi和TcpConfigurationSpi组合：


<?xml version="1.0" encoding="UTF-8"?>
 
<beans xmlns="http://www.springframework.org/schema/beans"
 
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 
xmlns:util="http://www.springframework.org/schema/util"
 
xsi:schemaLocation="
 
http://www.springframework.org/schema/beans
 
http://www.springframework.org/schema/beans/spring-beans.xsd
 
http://www.springframework.org/schema/util
 
http://www.springframework.org/schema/util/spring-util.xsd">
 
<bean id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
 
<property name="peerClassLoadingEnabled" value="true"/>
 
<property name="cacheConfiguration">
 
<list>
 
<!-- Partitioned cache example configuration (Atomic mode). -->
 
<bean class="org.apache.ignite.configuration.CacheConfiguration">
 
<property name="name" value="myCacheSource"/>
 
<property name="atomicityMode" value="ATOMIC"/>
 
<property name="backups" value="1"/>
 
</bean>
 
</list>
 
</property>
 
<!-- Enable cache events. -->
 
<property name="includeEventTypes">
 
<list>
 
<!-- Cache events. -->
 
<util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
 
</list>
 
</property>
 
<property name="discoverySpi">
 
<bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
 
<!-- Initial local port to listen to. -->
 
<property name="localPort" value="48500"/>
 
<!-- Changing local port range. This is an optional action. -->
 
<property name="localPortRange" value="20"/>
 
<!-- Setting up IP finder for this cluster -->
 
<property name="ipFinder">
 
<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
 
<property name="addresses">
 
<list>
 
<value>127.0.0.1:48500..48520</value>
 
</list>
 
</property>
 
</bean>
 
</property>
 
</bean>
 
</property>
 
<!--
 
Explicitly configure TCP communication SPI changing local
 
port number for the nodes from the first cluster.
 
-->
 
<property name="communicationSpi">
 
<bean class="org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi">
 
<property name="localPort" value="48100"/>
 
</bean>
 
</property>
 
</bean>
 
</beans>


我们指定了本地港口48500要侦听并使用静态IP查找器来发现节点，请执行以下操作。此外，我们还显式地将tcp通信端口配置为48100。从上述配置开始的每个Ignite节点将只加入到此集群，并且在同一主机上的另一个集群中不可见。注意，我们还启用了缓存对象放置用于获取的事件PUT缓存中每个条目的事件通知。作为数据源，我们将使用myCacheSource复制缓存。用名称保存文件isolated-cluster-1-kafka-source.xml在.。$IGNITE_HOME/examples/config文件夹。

对于来自第二个集群的节点，我们必须使用另一组端口。配置如下所示：


<?xml version="1.0" encoding="UTF-8"?>
 
<beans xmlns="http://www.springframework.org/schema/beans"
 
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 
xsi:schemaLocation="
 
http://www.springframework.org/schema/beans
 
http://www.springframework.org/schema/beans/spring-beans.xsd">
 
<bean id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
 
<property name="peerClassLoadingEnabled" value="true"/>
 
<property name="cacheConfiguration">
 
<list>
 
<!-- Partitioned cache example configuration (Atomic mode). -->
 
<bean class="org.apache.ignite.configuration.CacheConfiguration">
 
<property name="name" value="myCacheTarget"/>
 
<property name="atomicityMode" value="ATOMIC"/>
 
<property name="backups" value="1"/>
 
</bean>
 
</list>
 
</property>
 
<property name="discoverySpi">
 
<bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
 
<!-- Initial local port to listen to. -->
 
<property name="localPort" value="49500"/>
 
<!-- Changing local port range. This is an optional action. -->
 
<property name="localPortRange" value="20"/>
 
<!-- Setting up IP finder for this cluster -->
 
<property name="ipFinder">
 
<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
 
<property name="addresses">
 
<list>
 
<value>127.0.0.1:49500..49520</value>
 
</list>
 
</property>
 
</bean>
 
</property>
 
</bean>
 
</property>
 
<!--
 
Explicitly configure TCP communication SPI changing local port number
 
for the nodes from the second cluster.
 
-->
 
<property name="communicationSpi">
 
<bean class="org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi">
 
<property name="localPort" value="49100"/>
 
</bean>
 
</property>
 
</bean>
 
</beans>


对于第二个集群中的节点，我们将发现端口定义为49500和通信端口49100。这两种配置之间的差别是微不足道的，只有Spis和IP查找程序的端口号不同。将此配置保存为具有名称的文件isolated-cluster-1.xml并将文件放入文件夹中。$IGNITE_HOME/examples/config.

让我们测试一下配置。使用不同的配置文件在单独的控制台中启动两个Ignite节点。下面是一个如何运行Ignite节点的示例。


ignite.sh $IGNITE_HOME/examples/config/isolated-cluster-1-kafka-source.xml
 
ignite.sh $IGNITE_HOME/examples/config/isolated-cluster-2.xml


下一个屏幕截图显示了上述命令的结果。正如预期的那样，两个独立的Ignite节点在不同的集群中启动和运行。

通过Kafka在ignite集群之间进行实时数据复制_kafka_06

注意，所有清单和配置文件都可以在GitHub储存库.

第二步。接下来，您需要定义流提取器来将数据转换为键值元组。创建一个Maven项目，并将以下依赖项添加到pom.xml.


<dependency>
 
<groupId>org.apache.ignite</groupId>
 
<artifactId>ignite-kafka</artifactId>
 
<version>2.6.0</version>
 
</dependency>


我们用ignite-kafka模块作为我们的依赖项。添加以下具有名称的Java类CsvStreamExtractor进入com.blu.imdg包，它将实现StreamSingleTupleExtractor接口如下：


public class CsvStreamExtractor implements StreamSingleTupleExtractor<SinkRecord, String, String> {

 
public Map.Entry<String, String> extract(SinkRecord sinkRecord) {

 
System.out.println("SinkRecord:"+ sinkRecord.value().toString());
 
String[] parts = sinkRecord.value().toString().split(",");
 
String key = ((String[])parts[2].split("="))[1];
 
String val= ((String[])parts[7].split("="))[1];
 
return new AbstractMap.SimpleEntry<String, String>(key, val);
 
}
 
}


方法extract 是班里的工作人员吗？CsvStreamExtractor。这里的代码很简单：它从事件的每个元组中检索键和值，其中每个元组都公开为SinkRecord在小溪里。这，这个，那，那个extract 方法返回键值对，它将被发送到Ignite集群(目标)，以便在缓存中进一步存储。

使用Maven命令编译和构建项目：mvn clean install。成功编译项目后，一个名为kafka-1.0.jar应该在项目目标目录中创建。将库复制到文件夹$KAFKA_HOME/libs.

步骤3。现在我们的流提取器已经准备好使用了，让我们配置Ignite源和接收器连接器，并让它们从复制数据开始。让我们创建一个名为ignite-connector- source.properties进入$KAFKA_HOME/myconfig目录。添加以下属性并保存文件。


# connector
 
name=my-ignite-source-connector
 
connector.class=org.apache.ignite.stream.kafka.connect.IgniteSourceConnector
 
tasks.max=2
 
topicNames=test2
 
# cache
 
cacheName=myCacheSource
 
cacheAllowOverwrite=true
 
cacheEvts=put
 
igniteCfg=PATH_TO_THE_FILE/isolated-cluster-1-kafka-source.xml


在前面的连接器配置中，我们定义了org.apache.ignite.stream.kafka.connect.IgniteSourceConnector作为连接器类。我们还将test 2指定为主题名称，其中将存储流事件。接下来，对于缓存配置，我们将PUT事件定义为网格远程事件。在我们的例子中，我们使用myCacheSource作为源缓存。在这里，另一个关键属性是igniteCfg，其中我们显式地指定了一个孤立的集群配置。第一组将是我们的事件来源。

接下来，让我们配置Ignite接收器连接器。用名称创建另一个文件ignite- connector-sink.properties进入$KAFKA_HOME/myconfig目录。从下面的清单中添加以下属性。


# connector
 
name=my-ignite-sink-connector
 
connector.class=org.apache.ignite.stream.kafka.connect.IgniteSinkConnector
 
tasks.max=2
 
topics=test2
 
# cache
 
cacheName=myCacheTarget
 
cacheAllowOverwrite=true
 
igniteCfg=PATH_TO_THE_FILE/isolated-cluster-2.xml
 
singleTupleExtractorCls=com.blu.imdg.CsvStreamExtractor


配置与我们在上一节中使用的配置相同。主要区别在于singleTupleExtractorCls属性，其中指定了我们在步骤2中开发的流提取器。

步骤4。启动动物园管理员和Kafka Broker(服务器)，如Kafka文献. 

步骤5。您可以猜到，我们必须用test 2来创建一个新的Kafka主题。让我们使用以下命令创建主题。


$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor \ 1 --partitions 1 --topic test2


步骤6。让我们在一个单独的控制台中启动源和接收器连接器。首先，使用以下命令启动源连接器。


bin/connect-standalone.sh myconfig/connect-standalone.properties myconfig/ignite-connecto\ r-source.properties


这将使用默认连接器独立属性启动源连接器。请注意，此连接器还将启动Ignite服务器节点，我们将该节点加入到点燃团簇1.

拼图的最后一部分是沉槽连接器。我们现在已经准备好启动接收器连接器了。但是，在以独立模式启动另一个Kafka连接器之前，我们必须更改连接器的REST端口和存储文件名。创建一个名为Connection的文件-standalone-sink.properties进入$KAFKA_HOME/myconfig文件夹。向其添加以下属性。


bootstrap.servers=localhost:9092
 
key.converter=org.apache.kafka.connect.storage.StringConverter
 
value.converter=org.apache.kafka.connect.storage.StringConverter
 
key.converter.schemas.enable=false
 
value.converter.schemas.enable=false
 
internal.key.converter=org.apache.kafka.connect.storage.StringConverter
 
internal.value.converter=org.apache.kafka.connect.storage.StringConverter
 
internal.key.converter.schemas.enable=false
 
internal.value.converter.schemas.enable=false
 
offset.storage.file.filename=/tmp/connect-1.offsets
 
rest.port=8888
 
offset.flush.interval.ms=10000


大多数配置与以前相同，只有rest.port和offset.storage.file.filename是不同的。我们已经显式地定义了一个新的端口。8888对于此连接器，还指定了另一个文件存储。使用此配置从$KAFKA_HOME目录。


bin/connect-standalone.sh myconfig/connect-standalone-sink.properties myconfig/ignite-con\ nector-sink.properties


上面的命令将在另一个控制台上启动接收器连接器。下一个图显示了在独立控制台上启动和运行的两个连接器的屏幕截图。

通过Kafka在ignite集群之间进行实时数据复制_apache_07

步骤7。现在我们已经设置了连接器，现在是测试流管道的时候了。此时，如果我们在myCacheSource在集群1上创建的缓存，应该将条目复制到myCacheTarget集群2上的缓存。我们有几种方法可以使用IgniteRESTAPI或Java客户机将一些条目加载到缓存myCacheSource中。让我们使用Ignite Java客户机IsolatedCluster从…第二章写这篇文章的书。


$ java -jar ./target/IsolatedCLient-runnable.jar


这个Java客户机加载22进入缓存的条目myCacheSource。让我们观察一下在Ignite星系团上发生了什么。使用两个Ignite管理程序工具连接到集群，每个集群一个。执行cache -scan命令来扫描缓存，您应该得到一个非常类似于图5所示的截图。

通过Kafka在ignite集群之间进行实时数据复制_apache_08

如图5所示，不同集群中的每个缓存包含相同的条目集。如果仔细查看控制台上的Ignite接收器连接器日志，您应该会发现类似于以下日志的日志：


CacheEvent [cacheName=myCacheSource, part=64, key=Key:150, xid=null, lockId=GridCacheVersion [topVer=150300733, order=1538826349084, nodeOrder=4], newVal=Hello World!!: 150, oldVal=null, hasOldVal=false, hasNewVal=true, near=false, subjId=572ac224-f48b-4a0c-a844-496f4d609b6a, cloClsName=null, taskName=null, nodeId8=fb6ae4b6, evtNodeId8=572ac224, msg=Cache event., type=CACHE_OBJECT_PUT, tstamp=1538829094472]


Key:150


Val:Hello World!!:


启动源连接器流缓存。PUT事件进入主题test2作为一个元组，它包含元数据以及键和值：旧值和新值。点火器接收器连接器使用CsvStreamExtractor提取器从元组中检索值并将键值对存储到缓存中，myCacheTarget.

在上面的例子中，我们只配置了Ignite集群之间的单向实时数据复制。但是，ApacheIgniteKafka连接器将大量电源打包到一个小模块中。通过利用它的通用性和易用性，您可以开发强大的双向数据复制管道，或者在网格中发生任何缓存事件时通知任何客户端应用程序。此外，您还可以使用任何Kafka JDBC接收器连接器和Ignite源连接器将数据推入任何RDBMS。但是，Ignite源连接器也有一些限制，在生产环境中使用它之前应该考虑到这些限制：

  1. 点火源连接器不能并行工作。它不能分割工作，一个任务实例处理流。
  2. 它不处理多个缓存。为了处理多个缓存，必须定义在Kafka中配置和运行的多个连接器。
  3. 点火器源连接器需要在嵌入式模式下启动服务器节点才能获得通知的事件。
  4. 它不支持动态重新配置。

标签：ignite,Ignite,缓存,Kafka,集群,连接器,kafka
From： https://blog.51cto.com/u_16145034/6486162

相关文章

  • k8s集群根据进程PID获取Pod名称
    简单说明在实际的应用场景中，我们如果看到某个进程资源或服务异常，需要根据这个进程排查到底是哪个服务的Pod，这里我们介绍一种根据PID快速寻找Pod名称的方法。实际操作查看进程PID这里我们以GPU任务为例说明，可以看到占用显卡的任务PID为8241[root@centos~]#nvidia-smiThuJu......
  • Redis集群公网访问
    背景因业务需求，应用程序需要跨机房从公网地址访问Redis集群，但是无法正常访问。因为程序通过公网IP加端口访问到Redis集群，然后Redis返回集群信息（就是clusternodes命令的返回），程序再根据返回的集群信息去读写Redis集群。而当前集群监听在主机内网地址上，并且是通过内网地址创建的......
  • 一篇就让小白入门K8S，使用Minikube来搭建本地的单节点K8S集群
    开篇Kubernetes（通常简称为K8s）是一个开源的容器编排平台，它为应用程序的部署、扩展和管理提供了强大的工具和功能。在本文中，我们将通过一个实战案例，引导您进入Kubernetes的世界，了解其基本概念和架构，并帮助您安装和配置一个简单的Kubernetes集群。Kubernetes概述Kubernetes是一个......
  • 安装elasticsearch集群开启认证
    ##节点：172.30.1.101172.30.1.131172.30.1.102##下载rpmwgethttps://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.12.0-x86_64.rpm#安装elasticsearch##设置环境rpm-ivhelasticsearch-7.12.0-x86_64.rpm##禁用swapswapoff-ased-ri's/......
  • go实现高并发高可用分布式系统:设计类似kafka的高并发海量数据存储机制2
    上一节我们完成了数据的存储和索引，本节我们看如何写入数据和进行查询。我们将创建一个Segment对象，它一方面接收发送来的请求，也就是Record数据结构，然后将数据写入到store和index，基本架构如下：在前面章节中，我们使用代码定义了Record的数据结构，现在我们需要使用protobuf来重新定义它，一......
  • go实现高并发高可用分布式系统:设计类似kafka的高并发海量数据存储机制1
    上一节我们实现了日志微服务，它以http服务器的模式运行，客户端通过json方式将日志数据post过来，然后通过httpget的方式读取日志。当时我们的实现是将所有日志信息添加到数组末尾，这意味着所有日志信息都会保存在内存中。但分布式系统的日志数量将非常巨大，例如推特一天的日志数量就达到......
  • Redis集群
    Redis集群本章是基于CentOS7下的Redis集群教程，包括：单机安装RedisRedis主从Redis分片集群1.单机安装Redis首先需要安装Redis所需要的依赖：yuminstall-ygcctcl然后将课前资料提供的Redis安装包上传到虚拟机的任意目录：例如，我放到了/tmp目录：解压缩：tar-xvfredis-6......
  • mac下nginx+tomcat+memcached集群配置和共享session
     mac下nginx+tomcat+memcached集群配置和共享session  首先分别安装nginx（版本1.5.8）、tomcat（版本7.0）、memcached（版本1.4.17），具体安装说明请分别参考下面的文章。 nginx安装说明 http://stephen830.iteye.com/blog/2002195tomcat安装说明http://stephen830.iteye.com/blog/2001......
  • ubuntu2204脚本初始化集群
    @目录脚本实现脚本测试用于一键部署集群（多节点）的环境初始化，如修改主机名，主机映射，节点相互免密，时间同步等，可多节点部署ps：脚本需要下载sshpass，chrony等软件包，确保ubuntu源正常脚本实现#!/bin/bash#定义节点信息NODES=("192.168.200.180controllerroot""192.168.200.182c......
  • K8S集群中开启firewalld 防火墙，防火墙策略配置
    在所有节点上执行#确保开启防火墙服务systemctlrestartfirewalld#将集群内所有的节点IP配置到防火墙可信区中firewall-cmd--permanent--zone=trusted--add-source=172.17.185.91firewall-cmd--permanent--zone=trusted--add-source=172.17.185.92firewall-cmd-......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99