當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

实时数据交换平台 - BottledWater-pg with confluent

發布時間：2023/12/10 编程问答 42 豆豆

生活随笔收集整理的這篇文章主要介紹了实时数据交换平台 - BottledWater-pg with confluent 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

標簽

PostgreSQL , Bottled Water , Kafka , Confluent , IoT

背景

想必大家都在圖書館借過書，小時候有好看的書也會在小伙伴之間傳閱。

借書和數據泵有點類似，一份數據通過數據泵實時的分享給訂閱者。

例如在IoT的場景中，有流式分析的需求，也有存儲歷史數據的需求，同時還有數據挖掘的需求，搜索引擎可能也需要同一份數據，還有一些業務可能也要用到同一份數據。

但是如果把數據統統放到一個地方，這么多的業務，它們有的要求實時處理，有的要求批量處理，有的可能需要實時的更新數據，有的可能要對大數據進行分析。

顯然一個產品可能無法滿足這么多的需求。

就好比數據庫就分了關系數據庫，NOSQL，OLTP場景，OLAP場景一樣。也是因為一個產品無法滿足所有的業務需求。

在企業中通常是借助數據冗余來解決各類場景的需求。

那么如何才能夠更好的分享數據，保證數據的一致性，提高分享的實時性呢？

confluent platform

http://docs.confluent.io/3.1.0/platform.html

confluent 是一個實時的數據中轉服務，來自各個平臺的數據可以使用confluent進行流轉，達到分享和交換數據的目的。

例如來自物聯網傳感器的數據，來自數據庫的數據，來自HTTP，移動APP的數據，來自應用日志的數據，來自一些事件觸發的數據等等。

confluent需要依賴一些基本的組件，核心組件如kafka.

用戶可以自定義消息的生產者和消費者，在confluent提供的平臺上交換數據。

BottledWater-pg

bottledwater-pg是confluent平臺的一種消息生產者，針對PostgreSQL數據庫，即將PostgreSQL數據庫的數據寫入confluent Kafka，從而實時的分享給消息訂閱者。

支持PostgreSQL 9.4以及以上版本，支持全量快照，以及持續的增量數據寫入Kafka。

bottledwater-pg使用PostgreSQL快照技術，可以讀取一致性的快照寫入Kafka。使用數據庫logical decode技術，從PostgreSQL的WAL日志中，解析為ROW數據寫入Kafka。

在Kafka中，每個topic代表一張數據庫的表。

數據在使用decode從WAL取出后，寫入Kafka之前，使用Avro將數據ROW打包成JSON, or Protobuf, or Thrift, or any number of formats，再寫入Kafka。

Avro支持的數據類型比較豐富，可以很好的支撐PostgreSQL豐富的數據類型。

為什么使用Avro請參考

http://radar.oreilly.com/2014/11/the-problem-of-managing-schemas.html

BottledWater-pg依賴環境

BottledWater-pg是PG的一個插件，它的目的是解析WAL，同時使用Avro封裝為json/Protobuf/Thrift/其他formats。并寫入Kafka。

因此它依賴這些庫或軟件

PostgreSQL 9.4+ development libraries (PGXS and libpq). (Homebrew: brew install postgresql; Ubuntu: sudo apt-get install postgresql-server-dev-9.5 libpq-dev) libsnappy, a dependency of Avro. (Homebrew: brew install snappy; Ubuntu: sudo apt-get install libsnappy-dev) avro-c (1.8.0 or later), the C implementation of Avro. (Homebrew: brew install avro-c; others: build from source) Jansson, a JSON parser. (Homebrew: brew install jansson; Ubuntu: sudo apt-get install libjansson-dev) libcurl, a HTTP client. (Homebrew: brew install curl; Ubuntu: sudo apt-get install libcurl4-openssl-dev) librdkafka (0.9.1 or later), a Kafka client. (Ubuntu universe: sudo apt-get install librdkafka-dev, but see known gotchas; others: build from source)

部署BottledWater-pg

gcc,cmake

最好部署較新版本的，否則可能會有編譯問題。

gcc 6.2.0 python 2.7.12 cmake 3.6.3vi /etc/ld.so.conf /home/digoal/gcc6.2.0/lib /home/digoal/gcc6.2.0/lib64 /home/digoal/python2.7.12/libldconfigexport LD_LIBRARY_PATH=/home/digoal/gcc6.2.0/lib:/home/digoal/gcc6.2.0/lib64:/home/digoal/python2.7.12/lib:$LD_LIBRARY_PATH export PATH=/home/digoal/gcc6.2.0/bin:/home/digoal/python2.7.12/bin:/home/digoal/cmake3.6.3/bin:$PGHOME/bin:$PATH:.

snappy

可選，一種比較高效的壓縮和解壓縮庫。

由于avro還支持xz，可不安裝snappy

/* snappyhttp://google.github.io/snappy/wget https://github.com/google/snappy/archive/1.1.3.tar.gz tar -zxvf 1.1.3.tar.gzcd snappy-1.1.3yum install -y libtool gcc-c++ ./autogen.sh./configure --prefix=/home/digoal/snappy_home make make install- add LIBDIR to the `LD_LIBRARY_PATH' environment variableduring execution- add LIBDIR to the `LD_RUN_PATH' environment variableduring linking- use the `-Wl,-rpath -Wl,LIBDIR' linker flag- have your system administrator add LIBDIR to `/etc/ld.so.conf'*/

libjansson (libjansson >=2.3)

json parser，必須安裝，建議測試時安裝在默認目錄，否則可能遇到編譯問題，或者設置rpath。

http://www.digip.org/jansson/ wget http://www.digip.org/jansson/releases/jansson-2.9.tar.bz2tar -jxvf jansson-2.9.tar.bz2 cd jansson-2.9./configure --prefix=/home/digoal/jansson make make install- add LIBDIR to the 'LD_LIBRARY_PATH' environment variableduring execution- add LIBDIR to the 'LD_RUN_PATH' environment variableduring linking- use the '-Wl,-rpath -Wl,LIBDIR' linker flag- have your system administrator add LIBDIR to '/etc/ld.so.conf'export PKG_CONFIG_PATH=/home/digoal/jansson/lib/pkgconfig:$PKG_CONFIG_PATHpkg-config --cflags --libs jansson -I/home/digoal/jansson/include -L/home/digoal/jansson//home/digoal/jansson/lib -ljansson

建議測試時安裝在默認路徑中，如下。

./configure make make install export PKG_CONFIG_PATH=/usr/local/lib/pkgconfig:$PKG_CONFIG_PATH

liblzma

yum install -y xz-*

boost

可以不裝，如果你要安裝Avro的doc時才需要安裝boost。

/* boosthttp://www.boost.org/https://sourceforge.net/projects/boost/files/boost/1.62.0/ wget http://downloads.sourceforge.net/project/boost/boost/1.62.0/boost_1_62_0.tar.bz2?r=https%3A%2F%2Fsourceforge.net%2Fprojects%2Fboost%2Ffiles%2Fboost%2F1.62.0%2F&ts=1480929211&use_mirror=ncutar -jxvf boost_1_62_0.tar.bz2 cd boost_1_62_0/libs/regex/build如果要使用靜態庫，請執行make -fgcc.mak 如果要使用靜態庫，請執行make -fgcc-shared.makll gcc drwxr-xr-x 2 digoal users 4.0K Dec 5 17:18 boost_regex-gcc-1_53 drwxr-xr-x 2 digoal users 4.0K Dec 5 17:17 boost_regex-gcc-1_53_shared drwxr-xr-x 2 digoal users 4.0K Dec 5 17:19 boost_regex-gcc-d-1_53 drwxr-xr-x 2 digoal users 4.0K Dec 5 17:18 boost_regex-gcc-d-1_53_shared -rw-r--r-- 1 digoal users 2.6M Dec 5 17:18 libboost_regex-gcc-1_53.a -rwxr-xr-x 1 digoal users 1.3M Dec 5 17:17 libboost_regex-gcc-1_53.so -rw-r--r-- 1 digoal users 17M Dec 5 17:19 libboost_regex-gcc-d-1_53.a -rwxr-xr-x 1 digoal users 7.4M Dec 5 17:18 libboost_regex-gcc-d-1_53.solibboost_regex-gcc-1_53.a , 這是release版的靜態庫 libboost_regex-gcc-1_53.so , 這是release版的動態庫（共享庫） libboost_regex-gcc-d-1_53.a , 這是debug版的靜態庫 libboost_regex-gcc-d-1_53.so , 這里debug版的動態庫（共享庫）*/

avro (1.8.0 or later)

http://avro.apache.org/

http://www.apache.org/dyn/closer.cgi/avro/

wget http://mirrors.hust.edu.cn/apache/avro/avro-1.8.1/avro-src-1.8.1.tar.gztar -zxvf avro-src-1.8.1.tar.gzcd avro-src-1.8.1/lang/c mkdir build cd build/*cmake .. -DCMAKE_INSTALL_PREFIX=/home/digoal/avro -DCMAKE_BUILD_TYPE=Release -DSNAPPY_LIBRARIES=/home/digoal/snappy_home/lib -DSNAPPY_INCLUDE_DIR=/home/digoal/snappy_home/include */yum install -y zlib-devel.x86_64cmake .. -DCMAKE_INSTALL_PREFIX=/home/digoal/avro -DCMAKE_BUILD_TYPE=Release -DTHREADSAFE=true make make test make installThe "RelWithDebInfo" build type will build an optimized copy of the library, including debugging symbols. Use the "Release" build type if you don't want debugging symbols. Use the "Debug" build type if you want a non-optimized library, with debugging symbols.On Unix, you can request thread-safe versions of the Avro library's global functions by defining the THREADSAFE cmake variable. Just add the following to your cmake invokation:-DTHREADSAFE=true

libcurl

yum install -y libcurl-devel.x86_64

librdkafka

git clone https://github.com/edenhill/librdkafka /*./configure --prefix=/home/digoal/librdkafka_homemake -j 32 make installexport PKG_CONFIG_PATH=/home/digoal/avro/lib/pkgconfig:/home/digoal/librdkafka_home/lib/pkgconfig:/usr/local/lib/pkgconfig:$PKG_CONFIG_PATH*/

測試時建議按照在默認目錄，否則可能又會有編譯錯誤的問題。

./configure make make install export PKG_CONFIG_PATH=/usr/local/lib/pkgconfig:$PKG_CONFIG_PATH

PostgreSQL

安裝略

bottledwater-pg

由于bottledwater-pg是PostgreSQL的一個插件，所以首先要安裝PostgreSQL。

git clone https://github.com/confluentinc/bottledwater-pgcd bottledwater-pgmake make installvi /etc/ld.so.conf /home/digoal/avro/liborexport LD_LIBRARY_PATH=/home/digoal/avro/lib:$LD_LIBRARY_PATH

可能需要重啟數據庫，加載bottledwater.so

在數據庫中創建插件

psql postgres=# create extension bottledwater ; CREATE EXTENSION

部署confluent platform

這一步是基礎，搭建好confluent，后面才能測試一下bottledwater-pg生產消息，以及從confluent platform消費消息的過程。

http://docs.confluent.io/3.1.0/installation.html

部署PostgreSQL流復制

bottledwater-pg需要從wal解析logical row，所以必須配置WAL_LEVEL=logical級別。

同時wal sender進程數必須>=1。

worker process數必須>=1。

同時由于bottledwater為了保證可以支持斷點續傳，以及確保沒有轉換的WAL日志不會被主庫刪掉或覆蓋掉，需要用到replication slot，因此需要配置replication_slots>=1。

postgresql.confmax_worker_processes = 8 wal_level = logical max_wal_senders = 8 wal_keep_segments = 256 max_replication_slots = 4

同時為了保證數據庫的WAL訂閱者可以通過流復制協議連接到數據庫，需要配置pg_hba.conf

pg_hba.conflocal replication digoal trust host replication digoal 127.0.0.1/32 trust host replication digoal 0.0.0.0/0 md5

創建replication角色用戶

create role digoal login replication encrypted password 'digoal123';

使用bottledwater-pg生產消息

bottledwater-pg客戶端命令的目的是從WAL解析日志，寫入Kafka。

command option需要配置如何連接到數據庫（使用流復制連接），output格式，topic-prefix(建議為庫名)，是否需要初始化快照，是否允許沒有主鍵的表，kafka broker的連接地址和端口，schema-registry的連接地址和端口。

以及一些kafka相關的配置。

cd bottledwater-pg/kafka./bottledwater --help Exports a snapshot of a PostgreSQL database, followed by a stream of changes, and sends the data to a Kafka cluster.Usage:./bottledwater [OPTION]...Options:-d, --postgres=postgres://user:pass@host:port/dbname (required)Connection string or URI of the PostgreSQL server.-s, --slot=slotname Name of replication slot (default: bottledwater)The slot is automatically created on first use.-b, --broker=host1[:port1],host2[:port2]... (default: localhost:9092)Comma-separated list of Kafka broker hosts/ports.-r, --schema-registry=http://hostname:port (default: http://localhost:8081)URL of the service where Avro schemas are registered.Used only for --output-format=avro.Omit when --output-format=json.-f, --output-format=[avro|json] (default: avro)How to encode the messages for writing to Kafka.-u, --allow-unkeyed Allow export of tables that don't have a primary key.This is disallowed by default, because updates anddeletes need a primary key to identify their row.-p, --topic-prefix=prefixString to prepend to all topic names.e.g. with --topic-prefix=postgres, updates from table'users' will be written to topic 'postgres.users'.-e, --on-error=[log|exit] (default: exit)What to do in case of a transient error, such asfailure to publish to Kafka.-x, --skip-snapshot Skip taking a consistent snapshot of the existingdatabase contents and just start streaming any newupdates. (Ignored if the replication slot alreadyexists.)-C, --kafka-config property=valueSet global configuration property for Kafka producer(see --config-help for list of properties).-T, --topic-config property=valueSet topic configuration property for Kafka producer.--config-help Print the list of configuration properties. See also:https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md-h, --helpPrint this help text.

bottledwater配置文件說明如下

./bottledwater --config-help 2>&1 |less ## Global configuration propertiesProperty | C/P | Range | Default | Description -----------------------------------------|-----|-----------------|--------------:|-------------------------- builtin.features | * | | gzip, snappy, ssl, sasl, regex | Indicates the builtin features for this build of librdkafka. An application can either query this value or attempt to set it with its list of required features to check for library support. *Type: CSV flags* client.id | * | | rdkafka | Client identifier. *Type: string* metadata.broker.list | * | | | Initial list of brokers. The application may also use `rd_kafka_brokers_add()` to add brokers during runtime. *Type: string* bootstrap.servers | * | | | Alias for `metadata.broker.list` message.max.bytes | * | 1000 .. 1000000000 | 1000000 | Maximum transmit message size. *Type: integer* message.copy.max.bytes | * | 0 .. 1000000000 | 65535 | Maximum size for message to be copied to buffer. Messages larger than this will be passed by reference (zero-copy) at the expense of larger iovecs. *Type: integer* receive.message.max.bytes | * | 1000 .. 1000000000 | 100000000 | Maximum receive message size. This is a safety precaution to avoid memory exhaustion in case of protocol hickups. The value should be at least fetch.message.max.bytes * number of partitions consumed from + messaging overhead (e.g. 200000 bytes). *Type: integer* max.in.flight.requests.per.connection | * | 1 .. 1000000 | 1000000 | Maximum number of in-flight requests the client will send. This setting applies per broker connection. *Type: integer* max.in.flight | * | | | Alias for `max.in.flight.requests.per.connection` metadata.request.timeout.ms | * | 10 .. 900000 | 60000 | Non-topic request timeout in milliseconds. This is for metadata requests, etc. *Type: integer* topic.metadata.refresh.interval.ms | * | -1 .. 3600000 | 300000 | Topic metadata refresh interval in milliseconds. The metadata is automatically refreshed on error and connect. Use -1 to disable the intervalled refresh. *Type: integer* metadata.max.age.ms | * | | | Alias for `topic.metadata.refresh.interval.ms` topic.metadata.refresh.fast.cnt | * | 0 .. 1000 | 10 | When a topic looses its leader this number of metadata requests are sent with `topic.metadata.refresh.fast.interval.ms` interval disregarding the `topic.metadata.refresh.interval.ms` value. This is used to recover quickly from transitioning leader brokers. *Type: integer* topic.metadata.refresh.fast.interval.ms | * | 1 .. 60000 | 250 | See `topic.metadata.refresh.fast.cnt` description *Type: integer* topic.metadata.refresh.sparse | * | true, false | true | Sparse metadata requests (consumes less network bandwidth) *Type: boolean* topic.blacklist | * | | | Topic blacklist, a comma-separated list of regular expressions for matching topic names that should be ignored in broker metadata information as if the topics did not exist. *Type: pattern list* debug | * | generic, broker, topic, metadata, queue, msg, protocol, cgrp, security, fetch, feature, all | | A comma-separated list of debug contexts to enable. Debugging the Producer: broker,topic,msg. Consumer: cgrp,topic,fetch *Type: CSV flags* socket.timeout.ms | * | 10 .. 300000 | 60000 | Timeout for network requests. *Type: integer* socket.blocking.max.ms | * | 1 .. 60000 | 100 | Maximum time a broker socket operation may block. A lower value improves responsiveness at the expense of slightly higher CPU usage. *Type: integer* socket.send.buffer.bytes | * | 0 .. 100000000 | 0 | Broker socket send buffer size. System default is used if 0. *Type: integer* socket.receive.buffer.bytes | * | 0 .. 100000000 | 0 | Broker socket receive buffer size. System default is used if 0. *Type: integer* socket.keepalive.enable | * | true, false | false | Enable TCP keep-alives (SO_KEEPALIVE) on broker sockets *Type: boolean* socket.nagle.disable | * | true, false | false | Disable the Nagle algorithm (TCP_NODELAY). *Type: boolean* socket.max.fails | * | 0 .. 1000000 | 3 | Disconnect from broker when this number of send failures (e.g., timed out requests) is reached. Disable with 0. NOTE: The connection is automatically re-established. *Type: integer* broker.address.ttl | * | 0 .. 86400000 | 1000 | How long to cache the broker address resolving results (milliseconds). *Type: integer* broker.address.family | * | any, v4, v6 | any | Allowed broker IP address families: any, v4, v6 *Type: enum value* reconnect.backoff.jitter.ms | * | 0 .. 3600000 | 500 | Throttle broker reconnection attempts by this value +-50%. *Type: integer* statistics.interval.ms | * | 0 .. 86400000 | 0 | librdkafka statistics emit interval. The application also needs to register a stats callback using `rd_kafka_conf_set_stats_cb()`. The granularity is 1000ms. A value of 0 disables statistics. *Type: integer* enabled_events | * | 0 .. 2147483647 | 0 | See `rd_kafka_conf_set_events()` *Type: integer* error_cb | * | | | Error callback (set with rd_kafka_conf_set_error_cb()) *Type: pointer* throttle_cb | * | | | Throttle callback (set with rd_kafka_conf_set_throttle_cb()) *Type: pointer* stats_cb | * | | | Statistics callback (set with rd_kafka_conf_set_stats_cb()) *Type: pointer* log_cb | * | | | Log callback (set with rd_kafka_conf_set_log_cb()) *Type: pointer* log_level | * | 0 .. 7 | 6 | Logging level (syslog(3) levels) *Type: integer* log.thread.name | * | true, false | false | Print internal thread name in log messages (useful for debugging librdkafka internals) *Type: boolean* log.connection.close | * | true, false | true | Log broker disconnects. It might be useful to turn this off when interacting with 0.9 brokers with an aggressive `connection.max.idle.ms` value. *Type: boolean* socket_cb | * | | | Socket creation callback to provide race-free CLOEXEC *Type: pointer* connect_cb | * | | | Socket connect callback *Type: pointer* closesocket_cb | * | | | Socket close callback *Type: pointer* open_cb | * | | | File open callback to provide race-free CLOEXEC *Type: pointer* opaque | * | | | Application opaque (set with rd_kafka_conf_set_opaque()) *Type: pointer* default_topic_conf | * | | | Default topic configuration for automatically subscribed topics *Type: pointer* internal.termination.signal | * | 0 .. 128 | 0 | Signal that librdkafka will use to quickly terminate on rd_kafka_destroy(). If this signal is not set then there will be a delay before rd_kafka_wait_destroyed() returns true as internal threads are timing out their system calls. If this signal is set however the delay will be minimal. The application should mask this signal as an internal signal handler is installed. *Type: integer* api.version.request | * | true, false | false | Request broker's supported API versions to adjust functionality to available protocol features. If set to false the fallback version `broker.version.fallback` will be used. **NOTE**: Depends on broker version >=0.10.0. If the request is not supported by (an older) broker the `broker.version.fallback` fallback is used. *Type: boolean* api.version.fallback.ms | * | 0 .. 604800000 | 1200000 | Dictates how long the `broker.version.fallback` fallback is used in the case the ApiVersionRequest fails. **NOTE**: The ApiVersionRequest is only issued when a new connection to the broker is made (such as after an upgrade). *Type: integer* broker.version.fallback | * | | 0.9.0 | Older broker versions (<0.10.0) provides no way for a client to query for supported protocol features (ApiVersionRequest, see `api.version.request`) making it impossible for the client to know what features it may use. As a workaround a user may set this property to the expected broker version and the client will automatically adjust its feature set accordingly if the ApiVersionRequest fails (or is disabled). The fallback broker version will be used for `api.version.fallback.ms`. Valid values are: 0.9.0, 0.8.2, 0.8.1, 0.8.0. *Type: string* security.protocol | * | plaintext, ssl, sasl_plaintext, sasl_ssl | plaintext | Protocol used to communicate with brokers. *Type: enum value* ssl.cipher.suites | * | | | A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. See manual page for `ciphers(1)` and `SSL_CTX_set_cipher_list(3). *Type: string* ssl.key.location | * | | | Path to client's private key (PEM) used for authentication. *Type: string* ssl.key.password | * | | | Private key passphrase *Type: string* ssl.certificate.location | * | | | Path to client's public key (PEM) used for authentication. *Type: string* ssl.ca.location | * | | | File or directory path to CA certificate(s) for verifying the broker's key. *Type: string* ssl.crl.location | * | | | Path to CRL for verifying broker's certificate validity. *Type: string* sasl.mechanisms | * | | GSSAPI | SASL mechanism to use for authentication. Supported: GSSAPI, PLAIN. **NOTE**: Despite the name only one mechanism must be configured. *Type: string* sasl.kerberos.service.name | * | | kafka | Kerberos principal name that Kafka runs as. *Type: string* sasl.kerberos.principal | * | | kafkaclient | This client's Kerberos principal name. *Type: string* sasl.kerberos.kinit.cmd | * | | kinit -S "%{sasl.kerberos.service.name}/%{broker.name}" -k -t "%{sasl.kerberos.keytab}" %{sasl.kerberos.principal} | Full kerberos kinit command string, %{config.prop.name} is replaced by corresponding config object value, %{broker.name} returns the broker's hostname. *Type: string* sasl.kerberos.keytab | * | | | Path to Kerberos keytab file. Uses system default if not set.**NOTE**: This is not automatically used but must be added to the template in sasl.kerberos.kinit.cmd as ` ... -t %{sasl.kerberos.keytab}`. *Type: string* sasl.kerberos.min.time.before.relogin | * | 1 .. 86400000 | 60000 | Minimum time in milliseconds between key refresh attempts. *Type: integer* sasl.username | * | | | SASL username for use with the PLAIN mechanism *Type: string* sasl.password | * | | | SASL password for use with the PLAIN mechanism *Type: string* group.id | * | | | Client group id string. All clients sharing the same group.id belong to the same group. *Type: string* partition.assignment.strategy | * | | range,roundrobin | Name of partition assignment strategy to use when elected group leader assigns partitions to group members. *Type: string* session.timeout.ms | * | 1 .. 3600000 | 30000 | Client group session and failure detection timeout. *Type: integer* heartbeat.interval.ms | * | 1 .. 3600000 | 1000 | Group session keepalive heartbeat interval. *Type: integer* group.protocol.type | * | | consumer | Group protocol type *Type: string* coordinator.query.interval.ms | * | 1 .. 3600000 | 600000 | How often to query for the current client group coordinator. If the currently assigned coordinator is down the configured query interval will be divided by ten to more quickly recover in case of coordinator reassignment. *Type: integer* enable.auto.commit | C | true, false | true | Automatically and periodically commit offsets in the background. *Type: boolean* auto.commit.interval.ms | C | 0 .. 86400000 | 5000 | The frequency in milliseconds that the consumer offsets are committed (written) to offset storage. (0 = disable) *Type: integer* enable.auto.offset.store | C | true, false | true | Automatically store offset of last message provided to application. *Type: boolean* queued.min.messages | C | 1 .. 10000000 | 100000 | Minimum number of messages per topic+partition in the local consumer queue. *Type: integer* queued.max.messages.kbytes | C | 1 .. 1000000000 | 1000000 | Maximum number of kilobytes per topic+partition in the local consumer queue. This value may be overshot by fetch.message.max.bytes. *Type: integer* fetch.wait.max.ms | C | 0 .. 300000 | 100 | Maximum time the broker may wait to fill the response with fetch.min.bytes. *Type: integer* fetch.message.max.bytes | C | 1 .. 1000000000 | 1048576 | Initial maximum number of bytes per topic+partition to request when fetching messages from the broker. If the client encounters a message larger than this value it will gradually try to increase it until the entire message can be fetched. *Type: integer* max.partition.fetch.bytes | C | | | Alias for `fetch.message.max.bytes` fetch.min.bytes | C | 1 .. 100000000 | 1 | Minimum number of bytes the broker responds with. If fetch.wait.max.ms expires the accumulated data will be sent to the client regardless of this setting. *Type: integer* fetch.error.backoff.ms | C | 0 .. 300000 | 500 | How long to postpone the next fetch request for a topic+partition in case of a fetch error. *Type: integer* offset.store.method | C | none, file, broker | broker | Offset commit store method: 'file' - local file store (offset.store.path, et.al), 'broker' - broker commit store (requires Apache Kafka 0.8.2 or later on the broker). *Type: enum value* consume_cb | C | | | Message consume callback (set with rd_kafka_conf_set_consume_cb()) *Type: pointer* rebalance_cb | C | | | Called after consumer group has been rebalanced (set with rd_kafka_conf_set_rebalance_cb()) *Type: pointer* offset_commit_cb | C | | | Offset commit result propagation callback. (set with rd_kafka_conf_set_offset_commit_cb()) *Type: pointer* enable.partition.eof | C | true, false | true | Emit RD_KAFKA_RESP_ERR__PARTITION_EOF event whenever the consumer reaches the end of a partition. *Type: boolean* queue.buffering.max.messages | P | 1 .. 10000000 | 100000 | Maximum number of messages allowed on the producer queue. *Type: integer* queue.buffering.max.kbytes | P | 1 .. 2147483647 | 4000000 | Maximum total message size sum allowed on the producer queue. *Type: integer* queue.buffering.max.ms | P | 1 .. 900000 | 1000 | Maximum time, in milliseconds, for buffering data on the producer queue. *Type: integer* message.send.max.retries | P | 0 .. 10000000 | 2 | How many times to retry sending a failing MessageSet. **Note:** retrying may cause reordering. *Type: integer* retries | P | | | Alias for `message.send.max.retries` retry.backoff.ms | P | 1 .. 300000 | 100 | The backoff time in milliseconds before retrying a message send. *Type: integer* compression.codec | P | none, gzip, snappy | none | compression codec to use for compressing message sets. This is the default value for all topics, may be overriden by the topic configuration property `compression.codec`. *Type: enum value* batch.num.messages | P | 1 .. 1000000 | 10000 | Maximum number of messages batched in one MessageSet. The total MessageSet size is also limited by message.max.bytes. *Type: integer* delivery.report.only.error | P | true, false | false | Only provide delivery reports for failed messages. *Type: boolean* dr_cb | P | | | Delivery report callback (set with rd_kafka_conf_set_dr_cb()) *Type: pointer* dr_msg_cb | P | | | Delivery report callback (set with rd_kafka_conf_set_dr_msg_cb()) *Type: pointer*## Topic configuration propertiesProperty | C/P | Range | Default | Description -----------------------------------------|-----|-----------------|--------------:|-------------------------- request.required.acks | P | -1 .. 1000 | 1 | This field indicates how many acknowledgements the leader broker must receive from ISR brokers before responding to the request: *0*=Broker does not send any response/ack to client, *1*=Only the leader broker will need to ack the message, *-1* or *all*=broker will block until message is committed by all in sync replicas (ISRs) or broker's `in.sync.replicas` setting before sending response. *Type: integer* acks | P | | | Alias for `request.required.acks` request.timeout.ms | P | 1 .. 900000 | 5000 | The ack timeout of the producer request in milliseconds. This value is only enforced by the broker and relies on `request.required.acks` being != 0. *Type: integer* message.timeout.ms | P | 0 .. 900000 | 300000 | Local message timeout. This value is only enforced locally and limits the time a produced message waits for successful delivery. A time of 0 is infinite. *Type: integer* produce.offset.report | P | true, false | false | Report offset of produced message back to application. The application must be use the `dr_msg_cb` to retrieve the offset from `rd_kafka_message_t.offset`. *Type: boolean* partitioner_cb | P | | | Partitioner callback (set with rd_kafka_topic_conf_set_partitioner_cb()) *Type: pointer* opaque | * | | | Application opaque (set with rd_kafka_topic_conf_set_opaque()) *Type: pointer* compression.codec | P | none, gzip, snappy, inherit | inherit | Compression codec to use for compressing message sets. *Type: enum value* auto.commit.enable | C | true, false | true | If true, periodically commit offset of the last message handed to the application. This committed offset will be used when the process restarts to pick up where it left off. If false, the application will have to call `rd_kafka_offset_store()` to store an offset (optional). **NOTE:** This property should only be used with the simple legacy consumer, when using the high-level KafkaConsumer the global `enable.auto.commit` property must be used instead. **NOTE:** There is currently no zookeeper integration, offsets will be written to broker or local file according to offset.store.method. *Type: boolean* enable.auto.commit | C | | | Alias for `auto.commit.enable` auto.commit.interval.ms | C | 10 .. 86400000 | 60000 | The frequency in milliseconds that the consumer offsets are committed (written) to offset storage. *Type: integer* auto.offset.reset | C | smallest, earliest, beginning, largest, latest, end, error | largest | Action to take when there is no initial offset in offset store or the desired offset is out of range: 'smallest','earliest' - automatically reset the offset to the smallest offset, 'largest','latest' - automatically reset the offset to the largest offset, 'error' - trigger an error which is retrieved by consuming messages and checking 'message->err'. *Type: enum value* offset.store.path | C | | . | Path to local file for storing offsets. If the path is a directory a filename will be automatically generated in that directory based on the topic and partition. *Type: string* offset.store.sync.interval.ms | C | -1 .. 86400000 | -1 | fsync() interval for the offset file, in milliseconds. Use -1 to disable syncing, and 0 for immediate sync after each write. *Type: integer* offset.store.method | C | file, broker | broker | Offset commit store method: 'file' - local file store (offset.store.path, et.al), 'broker' - broker commit store (requires "group.id" to be configured and Apache Kafka 0.8.2 or later on the broker.). *Type: enum value* consume.callback.max.messages | C | 0 .. 1000000 | 0 | Maximum number of messages to dispatch in one `rd_kafka_consume_callback*()` call (0 = unlimited) *Type: integer*### C/P legend: C = Consumer, P = Producer, * = both

消費消息

由于confluent中存儲的是avro封裝的binary格式，所以消費時，需要使用解析avro的消費者。

./bin/kafka-avro-console-consumer --topic test --zookeeper localhost:2181 \--property print.key=true

風險評估

1. 首次連接數據庫時，會自動創建slot，同時自動開始將快照數據寫入Kafka，如果數據庫很大，這個過程會很漫長。

2. 為了得到一致的數據，會開啟repeatable read的事務隔離級別，如果是9.6，并且配置了snapshot too old參數，可能導致快照拷貝失敗。

3. 由于是邏輯DECODE，被復制的表必須包含逐漸，或指定非空唯一約束列，作為復制時的KEY。

4. 增，刪，改在WAL中被解析為：

insert : key + full row delete : old.key update : old.key + new.key+full row

如果使用--allow-unkeyed跳過了主鍵，那么delete該表時，不會將任何數據寫入Kafka，插入和更新則將所有列發給Kafka。

5. DDL操作不會記錄到wal日志中，如果你需要將DDL也寫入Kafka怎么辦？

你可以使用event trigger，發生ddl時，將DDL封裝并寫入表中，然后這些表的DML會寫入Kafka，從而實現DDL的傳遞。

6. 如果要刪除生產者，務必刪除數據庫中對應的slot ，否則PostgreSQL會一直保留slot未讀取的日志。導致WAL目錄撐爆。

7. 如果數據庫產生的REDO沒有被及時的解析并寫入Kafka，可能導致未取走的數據庫的wal文件一直留在數據庫服務器，甚至導致數據庫空間撐爆。

請謹慎使用slot，同時請將監控做得健壯。

8. Kafka topic與table一一對應，命名規則如下

由于命名中只有三個部分?[topic_prefix].[postgres_schema_name].table_name?沒有考慮庫名，所以如果有多個數據庫時，建議配置top_prefix，和庫名對應即可。

For each table being streamed, Bottled Water publishes messages to a corresponding Kafka topic. The naming convention for topics is [topic_prefix].[postgres_schema_name].table_name:table_name is the name of the table in Postgres. postgres_schema_name is the name of the Postgres schema the table belongs to; this is omitted if the schema is "public" (the default schema under the default Postgres configuration). N.B. this requires the avro-c library to be at least version 0.8.0. topic_prefix is omitted by default, but may be configured via the --topic-prefix command-line option. A prefix is useful: to prevent name collisions with other topics, if the Kafka broker is also being used for other purposes besides Bottled Water. if you want to stream several databases into the same broker, using a separate Bottled Water instance with a different prefix for each database. to make it easier for a Kafka consumer to consume updates from all Postgres tables, by using a topic regex that matches the prefix. For example:with no prefix configured, a table named "users" in the public (default) schema would be streamed to a topic named "users". with --topic-prefix=bottledwater, a table named "transactions" in the "point-of-sale" schema would be streamed to a topic named "bottledwater.point-of-sale.transactions". (Support for namespaces in Kafka has been proposed that would replace this sort of ad-hoc prefixing, but it's still under discussion.)

參考

1.?https://www.confluent.io/blog/bottled-water-real-time-integration-of-postgresql-and-kafka/

2.?http://docs.confluent.io/3.1.0/platform.html

3.?https://github.com/confluentinc/bottledwater-pg/tree/master#building-from-source

4.?http://docs.confluent.io/3.0.1/quickstart.html

5.?https://www.postgresql.org/message-id/797DF957-CE33-407F-99DB-7C7125E37ACE@kleppmann.com

總結

以上是生活随笔為你收集整理的实时数据交换平台 - BottledWater-pg with confluent的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Leetcode 1559二维网格图中探
下一篇：目标跟踪 — MOSSE