ਪੇਸ਼ ਕਰ ਰਿਹਾ ਹਾਂ ਡੇਬੇਜ਼ੀਅਮ - ਅਪਾਚੇ ਕਾਫਕਾ ਲਈ ਸੀਡੀਸੀ

ਮੇਰੇ ਕੰਮ ਵਿੱਚ, ਮੈਂ ਅਕਸਰ ਨਵੇਂ ਤਕਨੀਕੀ ਹੱਲ / ਸੌਫਟਵੇਅਰ ਉਤਪਾਦਾਂ ਵਿੱਚ ਆਉਂਦਾ ਹਾਂ, ਜਿਸ ਬਾਰੇ ਜਾਣਕਾਰੀ ਰੂਸੀ ਬੋਲਣ ਵਾਲੇ ਇੰਟਰਨੈਟ ਤੇ ਬਹੁਤ ਘੱਟ ਹੈ. ਇਸ ਲੇਖ ਦੇ ਨਾਲ, ਮੈਂ ਆਪਣੇ ਹਾਲੀਆ ਅਭਿਆਸ ਤੋਂ ਇੱਕ ਉਦਾਹਰਣ ਦੇ ਨਾਲ ਇੱਕ ਅਜਿਹੇ ਪਾੜੇ ਨੂੰ ਭਰਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰਾਂਗਾ, ਜਦੋਂ ਮੈਨੂੰ ਡੀਬੇਜ਼ੀਅਮ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ ਇੱਕ ਕਾਫਕਾ ਕਲੱਸਟਰ ਵਿੱਚ ਦੋ ਪ੍ਰਸਿੱਧ DBMSs (PostgreSQL ਅਤੇ MongoDB) ਤੋਂ CDC ਇਵੈਂਟਾਂ ਨੂੰ ਭੇਜਣ ਲਈ ਸੈੱਟਅੱਪ ਕਰਨ ਦੀ ਲੋੜ ਸੀ। ਮੈਨੂੰ ਉਮੀਦ ਹੈ ਕਿ ਇਹ ਸਮੀਖਿਆ ਲੇਖ, ਜੋ ਕਿ ਕੀਤੇ ਗਏ ਕੰਮ ਦੇ ਨਤੀਜੇ ਵਜੋਂ ਪ੍ਰਗਟ ਹੋਇਆ ਹੈ, ਦੂਜਿਆਂ ਲਈ ਲਾਭਦਾਇਕ ਹੋਵੇਗਾ.

ਆਮ ਤੌਰ 'ਤੇ ਡੀਬੇਜ਼ੀਅਮ ਅਤੇ ਸੀਡੀਸੀ ਕੀ ਹੈ?

ਡੀਬੇਜ਼ੀਅਮ - ਸੀਡੀਸੀ ਸਾਫਟਵੇਅਰ ਸ਼੍ਰੇਣੀ ਦਾ ਪ੍ਰਤੀਨਿਧੀ (ਡਾਟਾ ਤਬਦੀਲੀ ਕੈਪਚਰ ਕਰੋ), ਜਾਂ ਹੋਰ ਸਪੱਸ਼ਟ ਤੌਰ 'ਤੇ, ਇਹ ਵੱਖ-ਵੱਖ DBMSs ਲਈ ਕਨੈਕਟਰਾਂ ਦਾ ਇੱਕ ਸਮੂਹ ਹੈ ਜੋ ਅਪਾਚੇ ਕਾਫਕਾ ਕਨੈਕਟ ਫਰੇਮਵਰਕ ਦੇ ਅਨੁਕੂਲ ਹਨ।

ਇਸ ਓਪਨ ਸੋਰਸ ਪ੍ਰੋਜੈਕਟ, ਅਪਾਚੇ ਲਾਈਸੈਂਸ v2.0 ਅਧੀਨ ਲਾਇਸੰਸਸ਼ੁਦਾ ਅਤੇ Red Hat ਦੁਆਰਾ ਸਪਾਂਸਰ ਕੀਤਾ ਗਿਆ। ਵਿਕਾਸ 2016 ਤੋਂ ਚੱਲ ਰਿਹਾ ਹੈ ਅਤੇ ਇਸ ਸਮੇਂ ਇਹ ਹੇਠਾਂ ਦਿੱਤੇ DBMS ਲਈ ਅਧਿਕਾਰਤ ਸਹਾਇਤਾ ਪ੍ਰਦਾਨ ਕਰਦਾ ਹੈ: MySQL, PostgreSQL, MongoDB, SQL ਸਰਵਰ। ਕੈਸੈਂਡਰਾ ਅਤੇ ਓਰੇਕਲ ਲਈ ਵੀ ਕਨੈਕਟਰ ਹਨ, ਪਰ ਉਹ ਵਰਤਮਾਨ ਵਿੱਚ "ਸ਼ੁਰੂਆਤੀ ਪਹੁੰਚ" ਸਥਿਤੀ ਵਿੱਚ ਹਨ, ਅਤੇ ਨਵੇਂ ਰੀਲੀਜ਼ ਪਿਛੜੇ ਅਨੁਕੂਲਤਾ ਦੀ ਗਰੰਟੀ ਨਹੀਂ ਦਿੰਦੇ ਹਨ।

ਜੇ ਅਸੀਂ ਸੀਡੀਸੀ ਦੀ ਤੁਲਨਾ ਰਵਾਇਤੀ ਪਹੁੰਚ ਨਾਲ ਕਰਦੇ ਹਾਂ (ਜਦੋਂ ਐਪਲੀਕੇਸ਼ਨ ਸਿੱਧੇ DBMS ਤੋਂ ਡੇਟਾ ਪੜ੍ਹਦੀ ਹੈ), ਤਾਂ ਇਸਦੇ ਮੁੱਖ ਫਾਇਦਿਆਂ ਵਿੱਚ ਘੱਟ ਲੇਟੈਂਸੀ, ਉੱਚ ਭਰੋਸੇਯੋਗਤਾ ਅਤੇ ਉਪਲਬਧਤਾ ਦੇ ਨਾਲ ਕਤਾਰ ਪੱਧਰ 'ਤੇ ਡੇਟਾ ਤਬਦੀਲੀ ਸਟ੍ਰੀਮਿੰਗ ਨੂੰ ਲਾਗੂ ਕਰਨਾ ਸ਼ਾਮਲ ਹੈ। ਆਖਰੀ ਦੋ ਪੁਆਇੰਟ CDC ਇਵੈਂਟਸ ਲਈ ਇੱਕ ਰਿਪੋਜ਼ਟਰੀ ਵਜੋਂ ਕਾਫਕਾ ਕਲੱਸਟਰ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਪ੍ਰਾਪਤ ਕੀਤੇ ਜਾਂਦੇ ਹਨ।

ਨਾਲ ਹੀ, ਫਾਇਦਿਆਂ ਵਿੱਚ ਇਹ ਤੱਥ ਸ਼ਾਮਲ ਹੈ ਕਿ ਇੱਕ ਸਿੰਗਲ ਮਾਡਲ ਦੀ ਵਰਤੋਂ ਇਵੈਂਟਾਂ ਨੂੰ ਸਟੋਰ ਕਰਨ ਲਈ ਕੀਤੀ ਜਾਂਦੀ ਹੈ, ਇਸਲਈ ਅੰਤਿਮ ਐਪਲੀਕੇਸ਼ਨ ਨੂੰ ਵੱਖ-ਵੱਖ DBMS ਨੂੰ ਚਲਾਉਣ ਦੀਆਂ ਸੂਖਮਤਾਵਾਂ ਬਾਰੇ ਚਿੰਤਾ ਕਰਨ ਦੀ ਲੋੜ ਨਹੀਂ ਹੈ।

ਅੰਤ ਵਿੱਚ, ਇੱਕ ਸੁਨੇਹਾ ਬ੍ਰੋਕਰ ਦੀ ਵਰਤੋਂ ਕਰਨ ਨਾਲ ਐਪਲੀਕੇਸ਼ਨਾਂ ਦੀ ਹਰੀਜੱਟਲ ਸਕੇਲਿੰਗ ਦੀ ਗੁੰਜਾਇਸ਼ ਖੁੱਲ੍ਹ ਜਾਂਦੀ ਹੈ ਜੋ ਡੇਟਾ ਵਿੱਚ ਤਬਦੀਲੀਆਂ ਨੂੰ ਟਰੈਕ ਕਰਦੇ ਹਨ। ਉਸੇ ਸਮੇਂ, ਡਾਟਾ ਸਰੋਤ 'ਤੇ ਪ੍ਰਭਾਵ ਨੂੰ ਘੱਟ ਕੀਤਾ ਜਾਂਦਾ ਹੈ, ਕਿਉਂਕਿ ਡੇਟਾ ਸਿੱਧੇ DBMS ਤੋਂ ਨਹੀਂ, ਪਰ ਕਾਫਕਾ ਕਲੱਸਟਰ ਤੋਂ ਪ੍ਰਾਪਤ ਕੀਤਾ ਜਾਂਦਾ ਹੈ।

ਡੇਬੇਜ਼ੀਅਮ ਆਰਕੀਟੈਕਚਰ ਬਾਰੇ

Debezium ਦੀ ਵਰਤੋਂ ਕਰਨਾ ਇਸ ਸਧਾਰਨ ਸਕੀਮ 'ਤੇ ਆਉਂਦਾ ਹੈ:

DBMS (ਡਾਟਾ ਸਰੋਤ ਵਜੋਂ) → ਕਾਫਕਾ ਕਨੈਕਟ ਵਿੱਚ ਕਨੈਕਟਰ → ਅਪਾਚੇ ਕਾਫਕਾ → ਖਪਤਕਾਰ

ਇੱਕ ਉਦਾਹਰਣ ਦੇ ਤੌਰ 'ਤੇ, ਮੈਂ ਪ੍ਰੋਜੈਕਟ ਵੈਬਸਾਈਟ ਤੋਂ ਇੱਕ ਚਿੱਤਰ ਦੇਵਾਂਗਾ:

ਹਾਲਾਂਕਿ, ਮੈਨੂੰ ਇਹ ਸਕੀਮ ਅਸਲ ਵਿੱਚ ਪਸੰਦ ਨਹੀਂ ਹੈ, ਕਿਉਂਕਿ ਅਜਿਹਾ ਲਗਦਾ ਹੈ ਕਿ ਸਿਰਫ ਇੱਕ ਸਿੰਕ ਕਨੈਕਟਰ ਸੰਭਵ ਹੈ.

ਵਾਸਤਵ ਵਿੱਚ, ਸਥਿਤੀ ਵੱਖਰੀ ਹੈ: ਤੁਹਾਡੀ ਡੇਟਾ ਝੀਲ ਨੂੰ ਭਰਨਾ (ਉਪਰੋਕਤ ਚਿੱਤਰ ਵਿੱਚ ਆਖਰੀ ਲਿੰਕ) Debezium ਦੀ ਵਰਤੋਂ ਕਰਨ ਦਾ ਇੱਕੋ ਇੱਕ ਤਰੀਕਾ ਨਹੀਂ ਹੈ। ਅਪਾਚੇ ਕਾਫਕਾ ਨੂੰ ਭੇਜੇ ਗਏ ਇਵੈਂਟਸ ਨੂੰ ਤੁਹਾਡੀਆਂ ਐਪਲੀਕੇਸ਼ਨਾਂ ਦੁਆਰਾ ਵੱਖ-ਵੱਖ ਸਥਿਤੀਆਂ ਨੂੰ ਹੱਲ ਕਰਨ ਲਈ ਵਰਤਿਆ ਜਾ ਸਕਦਾ ਹੈ। ਉਦਾਹਰਣ ਲਈ:

• ਕੈਸ਼ ਤੋਂ ਅਪ੍ਰਸੰਗਿਕ ਡੇਟਾ ਨੂੰ ਹਟਾਉਣਾ;
• ਸੂਚਨਾਵਾਂ ਭੇਜਣਾ;
• ਖੋਜ ਸੂਚਕਾਂਕ ਅੱਪਡੇਟ;
• ਕੁਝ ਕਿਸਮ ਦੇ ਆਡਿਟ ਲੌਗ;
• ...

ਜੇਕਰ ਤੁਹਾਡੇ ਕੋਲ ਜਾਵਾ ਐਪਲੀਕੇਸ਼ਨ ਹੈ ਅਤੇ ਕਾਫਕਾ ਕਲੱਸਟਰ ਦੀ ਵਰਤੋਂ ਕਰਨ ਦੀ ਕੋਈ ਲੋੜ/ਸੰਭਾਵਨਾ ਨਹੀਂ ਹੈ, ਤਾਂ ਇਸ ਰਾਹੀਂ ਕੰਮ ਕਰਨ ਦੀ ਸੰਭਾਵਨਾ ਵੀ ਹੈ ਏਮਬੈਡਡ ਕਨੈਕਟਰ. ਸਪੱਸ਼ਟ ਪਲੱਸ ਇਹ ਹੈ ਕਿ ਇਸਦੇ ਨਾਲ ਤੁਸੀਂ ਵਾਧੂ ਬੁਨਿਆਦੀ ਢਾਂਚੇ (ਇੱਕ ਕਨੈਕਟਰ ਅਤੇ ਕਾਫਕਾ ਦੇ ਰੂਪ ਵਿੱਚ) ਤੋਂ ਇਨਕਾਰ ਕਰ ਸਕਦੇ ਹੋ. ਹਾਲਾਂਕਿ, ਇਹ ਹੱਲ ਸੰਸਕਰਣ 1.1 ਤੋਂ ਬਰਤਰਫ਼ ਕੀਤਾ ਗਿਆ ਹੈ ਅਤੇ ਹੁਣ ਵਰਤੋਂ ਲਈ ਸਿਫਾਰਸ਼ ਨਹੀਂ ਕੀਤੀ ਗਈ ਹੈ (ਇਸ ਨੂੰ ਭਵਿੱਖ ਦੇ ਰੀਲੀਜ਼ਾਂ ਵਿੱਚ ਹਟਾਇਆ ਜਾ ਸਕਦਾ ਹੈ)।

ਇਹ ਲੇਖ ਡਿਵੈਲਪਰਾਂ ਦੁਆਰਾ ਸਿਫਾਰਸ਼ ਕੀਤੇ ਆਰਕੀਟੈਕਚਰ ਦੀ ਚਰਚਾ ਕਰੇਗਾ, ਜੋ ਕਿ ਨੁਕਸ ਸਹਿਣਸ਼ੀਲਤਾ ਅਤੇ ਮਾਪਯੋਗਤਾ ਪ੍ਰਦਾਨ ਕਰਦਾ ਹੈ।

ਕਨੈਕਟਰ ਸੰਰਚਨਾ

ਸਭ ਤੋਂ ਮਹੱਤਵਪੂਰਨ ਮੁੱਲ - ਡੇਟਾ - ਵਿੱਚ ਤਬਦੀਲੀਆਂ ਨੂੰ ਟਰੈਕ ਕਰਨਾ ਸ਼ੁਰੂ ਕਰਨ ਲਈ ਸਾਨੂੰ ਲੋੜ ਹੈ:

1. ਡੇਟਾ ਸ੍ਰੋਤ, ਜੋ ਕਿ ਵਰਜਨ 5.7, PostgreSQL 9.6+, MongoDB 3.2+ ਤੋਂ ਸ਼ੁਰੂ ਹੋ ਕੇ MySQL ਹੋ ਸਕਦਾ ਹੈ (ਪੂਰੀ ਸੂਚੀ);
2. ਅਪਾਚੇ ਕਾਫਕਾ ਕਲੱਸਟਰ
3. ਕਾਫਕਾ ਕਨੈਕਟ ਉਦਾਹਰਨ (ਵਰਜਨ 1.x, 2.x);
4. ਡਿਬੇਜ਼ੀਅਮ ਕਨੈਕਟਰ ਨੂੰ ਸੰਰਚਿਤ ਕੀਤਾ।

ਪਹਿਲੇ ਦੋ ਬਿੰਦੂਆਂ 'ਤੇ ਕੰਮ ਕਰੋ, ਯਾਨੀ. ਡੀਬੀਐਮਐਸ ਅਤੇ ਅਪਾਚੇ ਕਾਫਕਾ ਨੂੰ ਸਥਾਪਤ ਕਰਨ ਦੀ ਪ੍ਰਕਿਰਿਆ ਲੇਖ ਦੇ ਦਾਇਰੇ ਤੋਂ ਬਾਹਰ ਹੈ। ਹਾਲਾਂਕਿ, ਉਹਨਾਂ ਲਈ ਜੋ ਇੱਕ ਸੈਂਡਬੌਕਸ ਵਿੱਚ ਹਰ ਚੀਜ਼ ਨੂੰ ਤੈਨਾਤ ਕਰਨਾ ਚਾਹੁੰਦੇ ਹਨ, ਉਦਾਹਰਣਾਂ ਦੇ ਨਾਲ ਅਧਿਕਾਰਤ ਰਿਪੋਜ਼ਟਰੀ ਵਿੱਚ ਇੱਕ ਤਿਆਰ-ਬਣਾਇਆ ਹੈ docker-compose.yaml.

ਅਸੀਂ ਆਖਰੀ ਦੋ ਬਿੰਦੂਆਂ 'ਤੇ ਵਧੇਰੇ ਵਿਸਥਾਰ ਨਾਲ ਧਿਆਨ ਦੇਵਾਂਗੇ।

0. ਕਾਫਕਾ ਕਨੈਕਟ

ਇੱਥੇ ਅਤੇ ਬਾਅਦ ਵਿੱਚ ਲੇਖ ਵਿੱਚ, ਸਾਰੀਆਂ ਕੌਂਫਿਗਰੇਸ਼ਨ ਉਦਾਹਰਣਾਂ ਨੂੰ ਡੇਬੇਜ਼ੀਅਮ ਡਿਵੈਲਪਰਾਂ ਦੁਆਰਾ ਵੰਡੇ ਗਏ ਡੌਕਰ ਚਿੱਤਰ ਦੇ ਸੰਦਰਭ ਵਿੱਚ ਵਿਚਾਰਿਆ ਜਾਂਦਾ ਹੈ। ਇਸ ਵਿੱਚ ਸਾਰੀਆਂ ਲੋੜੀਂਦੀਆਂ ਪਲੱਗਇਨ ਫਾਈਲਾਂ (ਕਨੈਕਟਰ) ਸ਼ਾਮਲ ਹਨ ਅਤੇ ਵਾਤਾਵਰਣ ਵੇਰੀਏਬਲ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਕਾਫਕਾ ਕਨੈਕਟ ਸੰਰਚਨਾ ਪ੍ਰਦਾਨ ਕਰਦਾ ਹੈ।

ਜੇਕਰ ਤੁਸੀਂ ਕਨਫਲੂਐਂਟ ਤੋਂ ਕਾਫਕਾ ਕਨੈਕਟ ਦੀ ਵਰਤੋਂ ਕਰਨ ਦਾ ਇਰਾਦਾ ਰੱਖਦੇ ਹੋ, ਤਾਂ ਤੁਹਾਨੂੰ ਲੋੜੀਂਦੇ ਕਨੈਕਟਰਾਂ ਦੇ ਪਲੱਗਇਨਾਂ ਨੂੰ ਆਪਣੇ ਆਪ ਵਿੱਚ ਨਿਰਧਾਰਿਤ ਡਾਇਰੈਕਟਰੀ ਵਿੱਚ ਜੋੜਨਾ ਪਵੇਗਾ। plugin.path ਜਾਂ ਵਾਤਾਵਰਣ ਵੇਰੀਏਬਲ ਦੁਆਰਾ ਸੈੱਟ ਕਰੋ CLASSPATH. ਕਾਫਕਾ ਕਨੈਕਟ ਵਰਕਰ ਅਤੇ ਕਨੈਕਟਰਾਂ ਲਈ ਸੈਟਿੰਗਾਂ ਕੌਂਫਿਗਰੇਸ਼ਨ ਫਾਈਲਾਂ ਦੁਆਰਾ ਪਰਿਭਾਸ਼ਿਤ ਕੀਤੀਆਂ ਜਾਂਦੀਆਂ ਹਨ ਜੋ ਕਿ ਵਰਕਰ ਸਟਾਰਟ ਕਮਾਂਡ ਨੂੰ ਆਰਗੂਮੈਂਟ ਵਜੋਂ ਪਾਸ ਕੀਤੀਆਂ ਜਾਂਦੀਆਂ ਹਨ। ਵੇਰਵਿਆਂ ਲਈ ਵੇਖੋ ਦਸਤਾਵੇਜ਼.

ਕਨੈਕਟਰ ਸੰਸਕਰਣ ਵਿੱਚ ਡੇਬੀਜ਼ਮ ਸਥਾਪਤ ਕਰਨ ਦੀ ਪੂਰੀ ਪ੍ਰਕਿਰਿਆ ਦੋ ਪੜਾਵਾਂ ਵਿੱਚ ਕੀਤੀ ਜਾਂਦੀ ਹੈ। ਆਓ ਉਨ੍ਹਾਂ ਵਿੱਚੋਂ ਹਰੇਕ 'ਤੇ ਵਿਚਾਰ ਕਰੀਏ:

1. ਕਾਫਕਾ ਕਨੈਕਟ ਫਰੇਮਵਰਕ ਸਥਾਪਤ ਕਰਨਾ

ਅਪਾਚੇ ਕਾਫਕਾ ਕਲੱਸਟਰ ਵਿੱਚ ਡੇਟਾ ਨੂੰ ਸਟ੍ਰੀਮ ਕਰਨ ਲਈ, ਕਾਫਕਾ ਕਨੈਕਟ ਫਰੇਮਵਰਕ ਵਿੱਚ ਖਾਸ ਮਾਪਦੰਡ ਸੈੱਟ ਕੀਤੇ ਗਏ ਹਨ, ਜਿਵੇਂ ਕਿ:

• ਕਲੱਸਟਰ ਕਨੈਕਸ਼ਨ ਸੈਟਿੰਗਾਂ,
• ਵਿਸ਼ਿਆਂ ਦੇ ਨਾਮ ਜਿਨ੍ਹਾਂ ਵਿੱਚ ਕਨੈਕਟਰ ਦੀ ਸੰਰਚਨਾ ਖੁਦ ਸਟੋਰ ਕੀਤੀ ਜਾਵੇਗੀ,
• ਸਮੂਹ ਦਾ ਨਾਮ ਜਿਸ ਵਿੱਚ ਕਨੈਕਟਰ ਚੱਲ ਰਿਹਾ ਹੈ (ਡਿਸਟ੍ਰੀਬਿਊਟਡ ਮੋਡ ਦੀ ਵਰਤੋਂ ਕਰਨ ਦੇ ਮਾਮਲੇ ਵਿੱਚ)।

ਪ੍ਰੋਜੈਕਟ ਦਾ ਅਧਿਕਾਰਤ ਡੌਕਰ ਚਿੱਤਰ ਵਾਤਾਵਰਣ ਵੇਰੀਏਬਲ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਸੰਰਚਨਾ ਦਾ ਸਮਰਥਨ ਕਰਦਾ ਹੈ - ਇਹ ਉਹ ਹੈ ਜੋ ਅਸੀਂ ਵਰਤਾਂਗੇ। ਤਾਂ ਆਓ ਚਿੱਤਰ ਨੂੰ ਡਾਊਨਲੋਡ ਕਰੀਏ:

docker pull debezium/connect

ਕੁਨੈਕਟਰ ਨੂੰ ਚਲਾਉਣ ਲਈ ਲੋੜੀਂਦੇ ਵਾਤਾਵਰਣ ਵੇਰੀਏਬਲਾਂ ਦਾ ਘੱਟੋ-ਘੱਟ ਸੈੱਟ ਹੇਠਾਂ ਦਿੱਤਾ ਗਿਆ ਹੈ:

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 - ਕਲੱਸਟਰ ਮੈਂਬਰਾਂ ਦੀ ਪੂਰੀ ਸੂਚੀ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ ਕਾਫਕਾ ਕਲੱਸਟਰ ਸਰਵਰਾਂ ਦੀ ਸ਼ੁਰੂਆਤੀ ਸੂਚੀ;
• OFFSET_STORAGE_TOPIC=connector-offsets - ਸਥਿਤੀਆਂ ਨੂੰ ਸਟੋਰ ਕਰਨ ਲਈ ਇੱਕ ਵਿਸ਼ਾ ਜਿੱਥੇ ਕੁਨੈਕਟਰ ਵਰਤਮਾਨ ਵਿੱਚ ਸਥਿਤ ਹੈ;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status - ਕਨੈਕਟਰ ਦੀ ਸਥਿਤੀ ਅਤੇ ਇਸਦੇ ਕਾਰਜਾਂ ਨੂੰ ਸਟੋਰ ਕਰਨ ਲਈ ਇੱਕ ਵਿਸ਼ਾ;
• CONFIG_STORAGE_TOPIC=connector-config - ਕਨੈਕਟਰ ਕੌਂਫਿਗਰੇਸ਼ਨ ਡੇਟਾ ਅਤੇ ਇਸਦੇ ਕਾਰਜਾਂ ਨੂੰ ਸਟੋਰ ਕਰਨ ਲਈ ਇੱਕ ਵਿਸ਼ਾ;
• GROUP_ID=1 — ਵਰਕਰਾਂ ਦੇ ਸਮੂਹ ਦਾ ਪਛਾਣਕਰਤਾ ਜਿਸ 'ਤੇ ਕਨੈਕਟਰ ਟਾਸਕ ਨੂੰ ਚਲਾਇਆ ਜਾ ਸਕਦਾ ਹੈ; ਡਿਸਟ੍ਰੀਬਿਊਟ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਸਮੇਂ ਲੋੜੀਂਦਾ ਹੈ (ਵੰਡਿਆ) ਸ਼ਾਸਨ.

ਅਸੀਂ ਇਹਨਾਂ ਵੇਰੀਏਬਲਾਂ ਨਾਲ ਕੰਟੇਨਰ ਸ਼ੁਰੂ ਕਰਦੇ ਹਾਂ:

docker run 
  -e BOOTSTRAP_SERVERS='kafka-1:9092,kafka-2:9092,kafka-3:9092' 
  -e GROUP_ID=1 
  -e CONFIG_STORAGE_TOPIC=my_connect_configs 
  -e OFFSET_STORAGE_TOPIC=my_connect_offsets 
  -e STATUS_STORAGE_TOPIC=my_connect_statuses  debezium/connect:1.2

ਐਵਰੋ ਬਾਰੇ ਨੋਟ ਕਰੋ

ਡਿਫੌਲਟ ਰੂਪ ਵਿੱਚ, Debezium JSON ਫਾਰਮੈਟ ਵਿੱਚ ਡੇਟਾ ਲਿਖਦਾ ਹੈ, ਜੋ ਕਿ ਸੈਂਡਬੌਕਸ ਅਤੇ ਘੱਟ ਮਾਤਰਾ ਵਿੱਚ ਡੇਟਾ ਲਈ ਸਵੀਕਾਰਯੋਗ ਹੈ, ਪਰ ਭਾਰੀ ਲੋਡ ਕੀਤੇ ਡੇਟਾਬੇਸ ਵਿੱਚ ਇੱਕ ਸਮੱਸਿਆ ਹੋ ਸਕਦੀ ਹੈ। JSON ਕਨਵਰਟਰ ਦਾ ਇੱਕ ਵਿਕਲਪ ਹੈ ਵਰਤਦੇ ਹੋਏ ਸੁਨੇਹਿਆਂ ਨੂੰ ਲੜੀਬੱਧ ਕਰਨਾ Avro ਇੱਕ ਬਾਈਨਰੀ ਫਾਰਮੈਟ ਵਿੱਚ, ਜੋ ਅਪਾਚੇ ਕਾਫਕਾ ਵਿੱਚ I/O ਸਬ-ਸਿਸਟਮ ਉੱਤੇ ਲੋਡ ਨੂੰ ਘਟਾਉਂਦਾ ਹੈ।

Avro ਦੀ ਵਰਤੋਂ ਕਰਨ ਲਈ, ਤੁਹਾਨੂੰ ਇੱਕ ਵੱਖਰਾ ਤੈਨਾਤ ਕਰਨ ਦੀ ਲੋੜ ਹੈ ਸਕੀਮਾ-ਰਜਿਸਟਰੀ (ਸਕੀਮਾਂ ਨੂੰ ਸਟੋਰ ਕਰਨ ਲਈ)। ਕਨਵਰਟਰ ਲਈ ਵੇਰੀਏਬਲ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਾਈ ਦੇਣਗੇ:

name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: VALUE_CONVERTER   
value: io.confluent.connect.avro.AvroConverter

ਐਵਰੋ ਦੀ ਵਰਤੋਂ ਕਰਨ ਅਤੇ ਇਸਦੇ ਲਈ ਇੱਕ ਰਜਿਸਟਰੀ ਸਥਾਪਤ ਕਰਨ ਬਾਰੇ ਵੇਰਵੇ ਲੇਖ ਦੇ ਦਾਇਰੇ ਤੋਂ ਬਾਹਰ ਹਨ - ਅੱਗੇ, ਸਪਸ਼ਟਤਾ ਲਈ, ਅਸੀਂ JSON ਦੀ ਵਰਤੋਂ ਕਰਾਂਗੇ।

2. ਕਨੈਕਟਰ ਨੂੰ ਖੁਦ ਸੈੱਟ ਕਰਨਾ

ਹੁਣ ਤੁਸੀਂ ਸਿੱਧੇ ਕਨੈਕਟਰ ਦੀ ਸੰਰਚਨਾ 'ਤੇ ਜਾ ਸਕਦੇ ਹੋ, ਜੋ ਸਰੋਤ ਤੋਂ ਡਾਟਾ ਪੜ੍ਹੇਗਾ।

ਆਉ ਦੋ DBMS ਲਈ ਕਨੈਕਟਰਾਂ ਦੀ ਉਦਾਹਰਨ ਵੇਖੀਏ: PostgreSQL ਅਤੇ MongoDB, ਜਿਸ ਲਈ ਮੇਰੇ ਕੋਲ ਅਨੁਭਵ ਹੈ ਅਤੇ ਜਿਸ ਲਈ ਅੰਤਰ ਹਨ (ਹਾਲਾਂਕਿ ਛੋਟੇ, ਪਰ ਕੁਝ ਮਾਮਲਿਆਂ ਵਿੱਚ ਮਹੱਤਵਪੂਰਨ!)

ਸੰਰਚਨਾ ਦਾ ਵਰਣਨ JSON ਨੋਟੇਸ਼ਨ ਵਿੱਚ ਕੀਤਾ ਗਿਆ ਹੈ ਅਤੇ ਇੱਕ POST ਬੇਨਤੀ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਕਾਫਕਾ ਕਨੈਕਟ 'ਤੇ ਅੱਪਲੋਡ ਕੀਤਾ ਗਿਆ ਹੈ।

2.1 PostgreSQL

PostgreSQL ਲਈ ਕਨੈਕਟਰ ਕੌਂਫਿਗਰੇਸ਼ਨ ਦੀ ਉਦਾਹਰਨ:

{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "127.0.0.1",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "definitelynotpassword",
    "database.dbname" : "dbname",
    "database.server.name": "pg-dev",
    "table.include.list": "public.(.*)",
    "heartbeat.interval.ms": "5000",
    "slot.name": "dbname_debezium",
    "publication.name": "dbname_publication",
    "transforms": "AddPrefix",
    "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.AddPrefix.regex": "pg-dev.public.(.*)",
    "transforms.AddPrefix.replacement": "data.cdc.dbname"
  }
}

ਇਸ ਸੰਰਚਨਾ ਦੇ ਬਾਅਦ ਕਨੈਕਟਰ ਦੇ ਸੰਚਾਲਨ ਦਾ ਸਿਧਾਂਤ ਕਾਫ਼ੀ ਸਧਾਰਨ ਹੈ:

• ਪਹਿਲੀ ਸ਼ੁਰੂਆਤ 'ਤੇ, ਇਹ ਸੰਰਚਨਾ ਵਿੱਚ ਨਿਰਧਾਰਤ ਡੇਟਾਬੇਸ ਨਾਲ ਜੁੜਦਾ ਹੈ ਅਤੇ ਮੋਡ ਵਿੱਚ ਸ਼ੁਰੂ ਹੁੰਦਾ ਹੈ ਸ਼ੁਰੂਆਤੀ ਸਨੈਪਸ਼ਾਟ, ਕੰਡੀਸ਼ਨਲ ਦੇ ਨਾਲ ਪ੍ਰਾਪਤ ਡੇਟਾ ਦਾ ਸ਼ੁਰੂਆਤੀ ਸੈੱਟ ਕਾਫਕਾ ਨੂੰ ਭੇਜਣਾ SELECT * FROM table_name.
• ਸ਼ੁਰੂਆਤੀਕਰਣ ਪੂਰਾ ਹੋਣ ਤੋਂ ਬਾਅਦ, ਕਨੈਕਟਰ PostgreSQL WAL ਫਾਈਲਾਂ ਤੋਂ ਰੀਡਿੰਗ ਤਬਦੀਲੀਆਂ ਦੇ ਮੋਡ ਵਿੱਚ ਦਾਖਲ ਹੁੰਦਾ ਹੈ।

ਵਰਤੇ ਗਏ ਵਿਕਲਪਾਂ ਬਾਰੇ:

• name — ਕੁਨੈਕਟਰ ਦਾ ਨਾਮ ਜਿਸ ਲਈ ਹੇਠਾਂ ਵਰਣਿਤ ਸੰਰਚਨਾ ਵਰਤੀ ਗਈ ਹੈ; ਭਵਿੱਖ ਵਿੱਚ, ਇਹ ਨਾਮ ਕਾਫਕਾ ਕਨੈਕਟ REST API ਦੁਆਰਾ ਕਨੈਕਟਰ (ਜਿਵੇਂ ਕਿ ਸਥਿਤੀ ਨੂੰ ਵੇਖਣ / ਮੁੜ ਚਾਲੂ ਕਰਨ / ਸੰਰਚਨਾ ਨੂੰ ਅੱਪਡੇਟ ਕਰਨ ਲਈ) ਨਾਲ ਕੰਮ ਕਰਨ ਲਈ ਵਰਤਿਆ ਜਾਂਦਾ ਹੈ;
• connector.class — DBMS ਕਨੈਕਟਰ ਕਲਾਸ ਜੋ ਕੌਂਫਿਗਰ ਕੀਤੇ ਕੁਨੈਕਟਰ ਦੁਆਰਾ ਵਰਤੀ ਜਾਵੇਗੀ;
• plugin.name WAL ਫਾਈਲਾਂ ਤੋਂ ਡੇਟਾ ਦੀ ਲਾਜ਼ੀਕਲ ਡੀਕੋਡਿੰਗ ਲਈ ਪਲੱਗਇਨ ਦਾ ਨਾਮ ਹੈ। ਚੁਣਨ ਲਈ ਉਪਲਬਧ ਹੈ wal2json, decoderbuffs и pgoutput. ਪਹਿਲੇ ਦੋ ਨੂੰ DBMS ਵਿੱਚ ਉਚਿਤ ਐਕਸਟੈਂਸ਼ਨਾਂ ਦੀ ਸਥਾਪਨਾ ਦੀ ਲੋੜ ਹੁੰਦੀ ਹੈ, ਅਤੇ pgoutput PostgreSQL ਸੰਸਕਰਣ 10 ਅਤੇ ਉੱਚ ਲਈ ਵਾਧੂ ਹੇਰਾਫੇਰੀ ਦੀ ਲੋੜ ਨਹੀਂ ਹੈ;
• database.* — ਡਾਟਾਬੇਸ ਨਾਲ ਜੁੜਨ ਲਈ ਵਿਕਲਪ, ਜਿੱਥੇ database.server.name - ਕਾਫਕਾ ਕਲੱਸਟਰ ਵਿੱਚ ਵਿਸ਼ੇ ਦਾ ਨਾਮ ਬਣਾਉਣ ਲਈ ਵਰਤੇ ਜਾਂਦੇ PostgreSQL ਉਦਾਹਰਣ ਦਾ ਨਾਮ;
• table.include.list - ਟੇਬਲ ਦੀ ਇੱਕ ਸੂਚੀ ਜਿਸ ਵਿੱਚ ਅਸੀਂ ਤਬਦੀਲੀਆਂ ਨੂੰ ਟਰੈਕ ਕਰਨਾ ਚਾਹੁੰਦੇ ਹਾਂ; ਫਾਰਮੈਟ ਵਿੱਚ ਦਿੱਤਾ ਗਿਆ ਹੈ schema.table_name; ਨਾਲ ਇਕੱਠੇ ਨਹੀਂ ਵਰਤਿਆ ਜਾ ਸਕਦਾ table.exclude.list;
• heartbeat.interval.ms - ਅੰਤਰਾਲ (ਮਿਲੀਸਕਿੰਟ ਵਿੱਚ) ਜਿਸ ਨਾਲ ਕਨੈਕਟਰ ਇੱਕ ਖਾਸ ਵਿਸ਼ੇ 'ਤੇ ਦਿਲ ਦੀ ਧੜਕਣ ਦੇ ਸੁਨੇਹੇ ਭੇਜਦਾ ਹੈ;
• heartbeat.action.query - ਇੱਕ ਬੇਨਤੀ ਜੋ ਹਰ ਦਿਲ ਦੀ ਧੜਕਣ ਦਾ ਸੁਨੇਹਾ ਭੇਜਣ ਵੇਲੇ ਲਾਗੂ ਕੀਤੀ ਜਾਵੇਗੀ (ਵਿਕਲਪ ਸੰਸਕਰਣ 1.1 ਤੋਂ ਪ੍ਰਗਟ ਹੋਇਆ ਹੈ);
• slot.name — ਰਿਪਲੀਕੇਸ਼ਨ ਸਲਾਟ ਦਾ ਨਾਮ ਜੋ ਕੁਨੈਕਟਰ ਦੁਆਰਾ ਵਰਤਿਆ ਜਾਵੇਗਾ;
• publication.name - ਨਾਮ ਪ੍ਰਕਾਸ਼ਨ PostgreSQL ਵਿੱਚ ਜੋ ਕੁਨੈਕਟਰ ਵਰਤਦਾ ਹੈ। ਜੇਕਰ ਇਹ ਮੌਜੂਦ ਨਹੀਂ ਹੈ, ਤਾਂ Debezium ਇਸਨੂੰ ਬਣਾਉਣ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰੇਗਾ। ਜੇਕਰ ਉਪਭੋਗਤਾ ਜਿਸ ਦੇ ਅਧੀਨ ਕੁਨੈਕਸ਼ਨ ਬਣਾਇਆ ਗਿਆ ਹੈ, ਕੋਲ ਇਸ ਕਾਰਵਾਈ ਲਈ ਲੋੜੀਂਦੇ ਅਧਿਕਾਰ ਨਹੀਂ ਹਨ, ਤਾਂ ਕਨੈਕਟਰ ਇੱਕ ਗਲਤੀ ਨਾਲ ਬਾਹਰ ਆ ਜਾਵੇਗਾ;
• transforms ਇਹ ਨਿਰਧਾਰਤ ਕਰਦਾ ਹੈ ਕਿ ਨਿਸ਼ਾਨਾ ਵਿਸ਼ੇ ਦਾ ਨਾਮ ਕਿਵੇਂ ਬਦਲਣਾ ਹੈ:
  • transforms.AddPrefix.type ਦਰਸਾਉਂਦਾ ਹੈ ਕਿ ਅਸੀਂ ਨਿਯਮਤ ਸਮੀਕਰਨਾਂ ਦੀ ਵਰਤੋਂ ਕਰਾਂਗੇ;
  • transforms.AddPrefix.regex - ਮਾਸਕ ਜਿਸ ਦੁਆਰਾ ਟੀਚੇ ਦੇ ਵਿਸ਼ੇ ਦਾ ਨਾਮ ਮੁੜ ਪਰਿਭਾਸ਼ਿਤ ਕੀਤਾ ਗਿਆ ਹੈ;
  • transforms.AddPrefix.replacement - ਸਿੱਧਾ ਜੋ ਅਸੀਂ ਮੁੜ ਪਰਿਭਾਸ਼ਿਤ ਕਰਦੇ ਹਾਂ।

ਦਿਲ ਦੀ ਧੜਕਣ ਅਤੇ ਪਰਿਵਰਤਨ ਬਾਰੇ ਹੋਰ

ਮੂਲ ਰੂਪ ਵਿੱਚ, ਕਨੈਕਟਰ ਹਰੇਕ ਵਚਨਬੱਧ ਟ੍ਰਾਂਜੈਕਸ਼ਨ ਲਈ ਕਾਫਕਾ ਨੂੰ ਡੇਟਾ ਭੇਜਦਾ ਹੈ, ਅਤੇ ਸੇਵਾ ਵਿਸ਼ੇ 'ਤੇ ਆਪਣਾ LSN (ਲੌਗ ਸੀਕਵੈਂਸ ਨੰਬਰ) ਲਿਖਦਾ ਹੈ। offset. ਪਰ ਕੀ ਹੁੰਦਾ ਹੈ ਜੇਕਰ ਕਨੈਕਟਰ ਨੂੰ ਪੂਰੇ ਡੇਟਾਬੇਸ ਨੂੰ ਪੜ੍ਹਨ ਲਈ ਸੰਰਚਿਤ ਨਹੀਂ ਕੀਤਾ ਗਿਆ ਹੈ, ਪਰ ਇਸਦੇ ਟੇਬਲਾਂ ਦਾ ਸਿਰਫ ਇੱਕ ਹਿੱਸਾ (ਜਿਸ ਵਿੱਚ ਡੇਟਾ ਕਦੇ-ਕਦਾਈਂ ਅਪਡੇਟ ਕੀਤਾ ਜਾਂਦਾ ਹੈ)?

• ਕਨੈਕਟਰ WAL ਫਾਈਲਾਂ ਨੂੰ ਪੜ੍ਹੇਗਾ ਅਤੇ ਉਹਨਾਂ ਵਿੱਚ ਟ੍ਰਾਂਜੈਕਸ਼ਨ ਕਮਿਟਾਂ ਦਾ ਪਤਾ ਨਹੀਂ ਲਗਾਏਗਾ ਉਹਨਾਂ ਟੇਬਲਾਂ ਵਿੱਚ ਜੋ ਇਹ ਮਾਨੀਟਰ ਕਰਦਾ ਹੈ।
• ਇਸ ਲਈ, ਇਹ ਆਪਣੀ ਮੌਜੂਦਾ ਸਥਿਤੀ ਨੂੰ ਵਿਸ਼ੇ ਵਿੱਚ ਜਾਂ ਪ੍ਰਤੀਕ੍ਰਿਤੀ ਸਲਾਟ ਵਿੱਚ ਅਪਡੇਟ ਨਹੀਂ ਕਰੇਗਾ।
• ਇਹ, ਬਦਲੇ ਵਿੱਚ, WAL ਫਾਈਲਾਂ ਨੂੰ ਡਿਸਕ ਉੱਤੇ "ਸਟੱਕ" ਕਰਨ ਦਾ ਕਾਰਨ ਬਣ ਜਾਵੇਗਾ ਅਤੇ ਸੰਭਾਵਤ ਤੌਰ 'ਤੇ ਡਿਸਕ ਸਪੇਸ ਖਤਮ ਹੋ ਜਾਵੇਗਾ।

ਅਤੇ ਇੱਥੇ ਵਿਕਲਪ ਬਚਾਅ ਲਈ ਆਉਂਦੇ ਹਨ. heartbeat.interval.ms и heartbeat.action.query. ਜੋੜਿਆਂ ਵਿੱਚ ਇਹਨਾਂ ਵਿਕਲਪਾਂ ਦੀ ਵਰਤੋਂ ਕਰਨ ਨਾਲ ਹਰ ਵਾਰ ਜਦੋਂ ਦਿਲ ਦੀ ਧੜਕਣ ਦਾ ਸੁਨੇਹਾ ਭੇਜਿਆ ਜਾਂਦਾ ਹੈ ਤਾਂ ਇੱਕ ਵੱਖਰੀ ਸਾਰਣੀ ਵਿੱਚ ਡੇਟਾ ਬਦਲਣ ਦੀ ਬੇਨਤੀ ਨੂੰ ਲਾਗੂ ਕਰਨਾ ਸੰਭਵ ਹੋ ਜਾਂਦਾ ਹੈ। ਇਸ ਤਰ੍ਹਾਂ, LSN ਜਿਸ 'ਤੇ ਕਨੈਕਟਰ ਵਰਤਮਾਨ ਵਿੱਚ ਸਥਿਤ ਹੈ (ਰਿਪਲੀਕੇਸ਼ਨ ਸਲਾਟ ਵਿੱਚ) ਲਗਾਤਾਰ ਅੱਪਡੇਟ ਕੀਤਾ ਜਾਂਦਾ ਹੈ। ਇਹ DBMS ਨੂੰ WAL ਫਾਈਲਾਂ ਨੂੰ ਹਟਾਉਣ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ ਜਿਨ੍ਹਾਂ ਦੀ ਹੁਣ ਲੋੜ ਨਹੀਂ ਹੈ। ਵਿਕਲਪ ਕਿਵੇਂ ਕੰਮ ਕਰਦੇ ਹਨ ਇਸ ਬਾਰੇ ਹੋਰ ਜਾਣਕਾਰੀ ਲਈ, ਵੇਖੋ ਦਸਤਾਵੇਜ਼.

ਇਕ ਹੋਰ ਵਿਕਲਪ ਜੋ ਨਜ਼ਦੀਕੀ ਧਿਆਨ ਦੇ ਹੱਕਦਾਰ ਹੈ transforms. ਹਾਲਾਂਕਿ ਇਹ ਸਹੂਲਤ ਅਤੇ ਸੁੰਦਰਤਾ ਬਾਰੇ ਵਧੇਰੇ ਹੈ ...

ਮੂਲ ਰੂਪ ਵਿੱਚ, Debezium ਹੇਠ ਦਿੱਤੀ ਨਾਮਕਰਨ ਨੀਤੀ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਵਿਸ਼ੇ ਬਣਾਉਂਦਾ ਹੈ: serverName.schemaName.tableName. ਇਹ ਹਮੇਸ਼ਾ ਸੁਵਿਧਾਜਨਕ ਨਹੀਂ ਹੋ ਸਕਦਾ ਹੈ। ਵਿਕਲਪ transforms ਰੈਗੂਲਰ ਸਮੀਕਰਨਾਂ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ, ਤੁਸੀਂ ਟੇਬਲਾਂ ਦੀ ਇੱਕ ਸੂਚੀ ਨੂੰ ਪਰਿਭਾਸ਼ਿਤ ਕਰ ਸਕਦੇ ਹੋ ਜਿਨ੍ਹਾਂ ਦੇ ਇਵੈਂਟਾਂ ਨੂੰ ਇੱਕ ਖਾਸ ਨਾਮ ਦੇ ਨਾਲ ਇੱਕ ਵਿਸ਼ੇ ਤੇ ਰੂਟ ਕਰਨ ਦੀ ਲੋੜ ਹੁੰਦੀ ਹੈ।

ਸਾਡੀ ਸੰਰਚਨਾ ਵਿੱਚ ਧੰਨਵਾਦ transforms ਇਹ ਵਾਪਰਦਾ ਹੈ: ਟਰੈਕ ਕੀਤੇ ਡੇਟਾਬੇਸ ਤੋਂ ਸਾਰੇ CDC ਇਵੈਂਟ ਨਾਮ ਦੇ ਨਾਲ ਵਿਸ਼ੇ 'ਤੇ ਜਾਣਗੇ data.cdc.dbname. ਨਹੀਂ ਤਾਂ (ਇਹਨਾਂ ਸੈਟਿੰਗਾਂ ਤੋਂ ਬਿਨਾਂ), Debezium ਮੂਲ ਰੂਪ ਵਿੱਚ ਫਾਰਮ ਦੀ ਹਰੇਕ ਸਾਰਣੀ ਲਈ ਇੱਕ ਵਿਸ਼ਾ ਬਣਾਵੇਗਾ: pg-dev.public.<table_name>.

ਕਨੈਕਟਰ ਸੀਮਾਵਾਂ

PostgreSQL ਲਈ ਕਨੈਕਟਰ ਸੰਰਚਨਾ ਦੇ ਵਰਣਨ ਦੇ ਅੰਤ ਵਿੱਚ, ਇਹ ਇਸਦੇ ਕੰਮ ਦੀਆਂ ਹੇਠ ਲਿਖੀਆਂ ਵਿਸ਼ੇਸ਼ਤਾਵਾਂ / ਸੀਮਾਵਾਂ ਬਾਰੇ ਗੱਲ ਕਰਨ ਯੋਗ ਹੈ:

1. PostgreSQL ਲਈ ਕਨੈਕਟਰ ਕਾਰਜਸ਼ੀਲਤਾ ਲਾਜ਼ੀਕਲ ਡੀਕੋਡਿੰਗ ਦੀ ਧਾਰਨਾ 'ਤੇ ਨਿਰਭਰ ਕਰਦੀ ਹੈ। ਇਸ ਲਈ ਉਹ ਡੇਟਾਬੇਸ ਦੀ ਬਣਤਰ ਨੂੰ ਬਦਲਣ ਲਈ ਬੇਨਤੀਆਂ ਨੂੰ ਟਰੈਕ ਨਹੀਂ ਕਰਦਾ ਹੈ (DDL) - ਇਸਦੇ ਅਨੁਸਾਰ, ਇਹ ਡੇਟਾ ਵਿਸ਼ਿਆਂ ਵਿੱਚ ਨਹੀਂ ਹੋਵੇਗਾ.
2. ਕਿਉਂਕਿ ਪ੍ਰਤੀਕ੍ਰਿਤੀ ਸਲਾਟ ਵਰਤੇ ਜਾਂਦੇ ਹਨ, ਕਨੈਕਟਰ ਦਾ ਕੁਨੈਕਸ਼ਨ ਸੰਭਵ ਹੈ ਸਿਰਫ ਮਾਸਟਰ DBMS ਉਦਾਹਰਨ ਲਈ।
3. ਜੇਕਰ ਉਪਭੋਗਤਾ ਜਿਸ ਦੇ ਅਧੀਨ ਕਨੈਕਟਰ ਡੇਟਾਬੇਸ ਨਾਲ ਜੁੜਦਾ ਹੈ, ਕੋਲ ਸਿਰਫ਼-ਪੜ੍ਹਨ ਦੇ ਅਧਿਕਾਰ ਹਨ, ਤਾਂ ਪਹਿਲੇ ਲਾਂਚ ਤੋਂ ਪਹਿਲਾਂ, ਤੁਹਾਨੂੰ ਦਸਤੀ ਇੱਕ ਪ੍ਰਤੀਕ੍ਰਿਤੀ ਸਲਾਟ ਬਣਾਉਣ ਅਤੇ ਡੇਟਾਬੇਸ ਵਿੱਚ ਪ੍ਰਕਾਸ਼ਿਤ ਕਰਨ ਦੀ ਲੋੜ ਹੋਵੇਗੀ।

ਇੱਕ ਸੰਰਚਨਾ ਲਾਗੂ ਕੀਤੀ ਜਾ ਰਹੀ ਹੈ

ਤਾਂ ਆਉ ਆਪਣੀ ਸੰਰਚਨਾ ਨੂੰ ਕਨੈਕਟਰ ਵਿੱਚ ਲੋਡ ਕਰੀਏ:

curl -i -X POST -H "Accept:application/json" 
  -H  "Content-Type:application/json"  http://localhost:8083/connectors/ 
  -d @pg-con.json

ਅਸੀਂ ਜਾਂਚ ਕਰਦੇ ਹਾਂ ਕਿ ਡਾਊਨਲੋਡ ਸਫਲ ਸੀ ਅਤੇ ਕਨੈਕਟਰ ਸ਼ੁਰੂ ਹੋਇਆ:

$ curl -i http://localhost:8083/connectors/pg-connector/status 
HTTP/1.1 200 OK
Date: Thu, 17 Sep 2020 20:19:40 GMT
Content-Type: application/json
Content-Length: 175
Server: Jetty(9.4.20.v20190813)

{"name":"pg-connector","connector":{"state":"RUNNING","worker_id":"172.24.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.24.0.5:8083"}],"type":"source"}

ਸ਼ਾਨਦਾਰ: ਇਹ ਸੈੱਟਅੱਪ ਹੈ ਅਤੇ ਜਾਣ ਲਈ ਤਿਆਰ ਹੈ। ਆਉ ਹੁਣ ਇੱਕ ਖਪਤਕਾਰ ਹੋਣ ਦਾ ਦਿਖਾਵਾ ਕਰੀਏ ਅਤੇ ਕਾਫਕਾ ਨਾਲ ਜੁੜੀਏ, ਜਿਸ ਤੋਂ ਬਾਅਦ ਅਸੀਂ ਸਾਰਣੀ ਵਿੱਚ ਇੱਕ ਐਂਟਰੀ ਜੋੜਦੇ ਅਤੇ ਬਦਲਦੇ ਹਾਂ:

$ kafka/bin/kafka-console-consumer.sh 
  --bootstrap-server kafka:9092 
  --from-beginning 
  --property print.key=true 
  --topic data.cdc.dbname

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', '[email protected]');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

ਸਾਡੇ ਵਿਸ਼ੇ ਵਿੱਚ, ਇਹ ਇਸ ਤਰ੍ਹਾਂ ਪ੍ਰਦਰਸ਼ਿਤ ਹੋਵੇਗਾ:

ਸਾਡੇ ਬਦਲਾਅ ਦੇ ਨਾਲ ਬਹੁਤ ਲੰਮਾ JSON

{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":null,
"after":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600374991648,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":602,
"lsn":34088472,
"xmin":null
},
"op":"c",
"ts_ms":1600374991762,
"transaction":null
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"after":{
"id":1005,
"first_name":"egg",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600375609365,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":603,
"lsn":34089688,
"xmin":null
},
"op":"u",
"ts_ms":1600375609778,
"transaction":null
}
}

ਦੋਵਾਂ ਮਾਮਲਿਆਂ ਵਿੱਚ, ਰਿਕਾਰਡਾਂ ਵਿੱਚ ਉਸ ਰਿਕਾਰਡ ਦੀ ਕੁੰਜੀ (PK) ਸ਼ਾਮਲ ਹੁੰਦੀ ਹੈ ਜੋ ਬਦਲਿਆ ਗਿਆ ਸੀ, ਅਤੇ ਤਬਦੀਲੀਆਂ ਦਾ ਸਾਰ: ਰਿਕਾਰਡ ਪਹਿਲਾਂ ਕੀ ਸੀ ਅਤੇ ਬਾਅਦ ਵਿੱਚ ਕੀ ਬਣਿਆ।

• ਦੇ ਮਾਮਲੇ ਵਿਚ INSERT: ਪਹਿਲਾਂ ਮੁੱਲ (before) ਬਰਾਬਰ nullਉਸ ਤੋਂ ਬਾਅਦ ਸਤਰ ਜੋ ਪਾਈ ਗਈ ਸੀ।
• ਦੇ ਮਾਮਲੇ ਵਿਚ UPDATE: ਤੇ payload.before ਕਤਾਰ ਦੀ ਪਿਛਲੀ ਸਥਿਤੀ ਵੇਖਾਈ ਜਾਂਦੀ ਹੈ, ਅਤੇ ਵਿੱਚ payload.after - ਤਬਦੀਲੀ ਦੇ ਤੱਤ ਦੇ ਨਾਲ ਨਵਾਂ।

2.2 ਮੋਂਗੋਡੀਬੀ

ਇਹ ਕਨੈਕਟਰ ਮਿਆਰੀ MongoDB ਪ੍ਰਤੀਕ੍ਰਿਤੀ ਵਿਧੀ ਦੀ ਵਰਤੋਂ ਕਰਦਾ ਹੈ, DBMS ਪ੍ਰਾਇਮਰੀ ਨੋਡ ਦੇ ਓਪਲੌਗ ਤੋਂ ਜਾਣਕਾਰੀ ਪੜ੍ਹਦਾ ਹੈ।

ਇਸੇ ਤਰ੍ਹਾਂ PgSQL ਲਈ ਪਹਿਲਾਂ ਹੀ ਵਰਣਿਤ ਕਨੈਕਟਰ ਲਈ, ਇੱਥੇ ਵੀ, ਪਹਿਲੀ ਸ਼ੁਰੂਆਤ 'ਤੇ, ਪ੍ਰਾਇਮਰੀ ਡਾਟਾ ਸਨੈਪਸ਼ਾਟ ਲਿਆ ਜਾਂਦਾ ਹੈ, ਜਿਸ ਤੋਂ ਬਾਅਦ ਕਨੈਕਟਰ ਔਪਲੌਗ ਰੀਡਿੰਗ ਮੋਡ 'ਤੇ ਸਵਿਚ ਕਰਦਾ ਹੈ।

ਸੰਰਚਨਾ ਉਦਾਹਰਨ:

{
"name": "mp-k8s-mongo-connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"tasks.max": "1",
"mongodb.hosts": "MainRepSet/mongo:27017",
"mongodb.name": "mongo",
"mongodb.user": "debezium",
"mongodb.password": "dbname",
"database.whitelist": "db_1,db_2",
"transforms": "AddPrefix",
"transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex": "mongo.([a-zA-Z_0-9]*).([a-zA-Z_0-9]*)",
"transforms.AddPrefix.replacement": "data.cdc.mongo_$1"
}
}

ਜਿਵੇਂ ਕਿ ਤੁਸੀਂ ਵੇਖ ਸਕਦੇ ਹੋ, ਪਿਛਲੀ ਉਦਾਹਰਨ ਦੇ ਮੁਕਾਬਲੇ ਕੋਈ ਨਵੇਂ ਵਿਕਲਪ ਨਹੀਂ ਹਨ, ਪਰ ਡੇਟਾਬੇਸ ਨਾਲ ਜੁੜਨ ਲਈ ਜ਼ਿੰਮੇਵਾਰ ਵਿਕਲਪਾਂ ਦੀ ਗਿਣਤੀ ਅਤੇ ਉਹਨਾਂ ਦੇ ਅਗੇਤਰ ਨੂੰ ਘਟਾ ਦਿੱਤਾ ਗਿਆ ਹੈ।

ਸੈਟਿੰਗਾਂ transforms ਇਸ ਵਾਰ ਉਹ ਹੇਠਾਂ ਦਿੱਤੇ ਕੰਮ ਕਰਦੇ ਹਨ: ਸਕੀਮ ਤੋਂ ਟੀਚੇ ਦੇ ਵਿਸ਼ੇ ਦਾ ਨਾਮ ਬਦਲੋ <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

ਨੁਕਸ ਸਹਿਣਸ਼ੀਲਤਾ

ਸਾਡੇ ਸਮੇਂ ਵਿੱਚ ਨੁਕਸ ਸਹਿਣਸ਼ੀਲਤਾ ਅਤੇ ਉੱਚ ਉਪਲਬਧਤਾ ਦਾ ਮੁੱਦਾ ਪਹਿਲਾਂ ਨਾਲੋਂ ਵਧੇਰੇ ਗੰਭੀਰ ਹੈ - ਖਾਸ ਕਰਕੇ ਜਦੋਂ ਅਸੀਂ ਡੇਟਾ ਅਤੇ ਟ੍ਰਾਂਜੈਕਸ਼ਨਾਂ ਬਾਰੇ ਗੱਲ ਕਰਦੇ ਹਾਂ, ਅਤੇ ਡੇਟਾ ਪਰਿਵਰਤਨ ਟਰੈਕਿੰਗ ਇਸ ਮਾਮਲੇ ਵਿੱਚ ਪਾਸੇ ਨਹੀਂ ਹੈ। ਆਓ ਦੇਖੀਏ ਕਿ ਸਿਧਾਂਤਕ ਤੌਰ 'ਤੇ ਕੀ ਗਲਤ ਹੋ ਸਕਦਾ ਹੈ ਅਤੇ ਹਰੇਕ ਮਾਮਲੇ ਵਿੱਚ ਡੇਬੇਜ਼ੀਅਮ ਦਾ ਕੀ ਹੋਵੇਗਾ.

ਤਿੰਨ ਔਪਟ-ਆਊਟ ਵਿਕਲਪ ਹਨ:

1. ਕਾਫਕਾ ਕਨੈਕਟ ਅਸਫਲਤਾ. ਜੇਕਰ ਕਨੈਕਟ ਨੂੰ ਡਿਸਟਰੀਬਿਊਟਡ ਮੋਡ ਵਿੱਚ ਕੰਮ ਕਰਨ ਲਈ ਕੌਂਫਿਗਰ ਕੀਤਾ ਗਿਆ ਹੈ, ਤਾਂ ਇਸ ਲਈ ਇੱਕ ਤੋਂ ਵੱਧ ਕਰਮਚਾਰੀਆਂ ਨੂੰ ਇੱਕੋ group.id ਸੈੱਟ ਕਰਨ ਦੀ ਲੋੜ ਹੁੰਦੀ ਹੈ। ਫਿਰ, ਜੇਕਰ ਉਹਨਾਂ ਵਿੱਚੋਂ ਇੱਕ ਅਸਫਲ ਹੋ ਜਾਂਦਾ ਹੈ, ਤਾਂ ਕਨੈਕਟਰ ਨੂੰ ਦੂਜੇ ਕਰਮਚਾਰੀ 'ਤੇ ਮੁੜ ਚਾਲੂ ਕੀਤਾ ਜਾਵੇਗਾ ਅਤੇ ਕਾਫਕਾ ਵਿੱਚ ਵਿਸ਼ੇ ਵਿੱਚ ਆਖਰੀ ਪ੍ਰਤੀਬੱਧ ਸਥਿਤੀ ਤੋਂ ਪੜ੍ਹਨਾ ਜਾਰੀ ਰੱਖਿਆ ਜਾਵੇਗਾ।
2. ਕਾਫਕਾ ਕਲੱਸਟਰ ਨਾਲ ਸੰਪਰਕ ਦਾ ਨੁਕਸਾਨ. ਕਨੈਕਟਰ ਸਿਰਫ਼ ਉਸ ਸਥਿਤੀ 'ਤੇ ਪੜ੍ਹਨਾ ਬੰਦ ਕਰ ਦੇਵੇਗਾ ਜਿਸ ਨੂੰ ਇਹ ਕਾਫਕਾ ਨੂੰ ਭੇਜਣ ਵਿੱਚ ਅਸਫਲ ਰਿਹਾ ਹੈ ਅਤੇ ਸਮੇਂ-ਸਮੇਂ 'ਤੇ ਇਸਨੂੰ ਦੁਬਾਰਾ ਭੇਜਣ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰੇਗਾ ਜਦੋਂ ਤੱਕ ਕੋਸ਼ਿਸ਼ ਸਫਲ ਨਹੀਂ ਹੋ ਜਾਂਦੀ।
3. ਡਾਟਾ ਸਰੋਤ ਉਪਲਬਧ ਨਹੀਂ ਹੈ. ਕਨੈਕਟਰ ਸੰਰਚਨਾ ਦੇ ਅਨੁਸਾਰ ਸਰੋਤ ਨਾਲ ਮੁੜ ਕਨੈਕਟ ਕਰਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰੇਗਾ। ਪੂਰਵ-ਨਿਰਧਾਰਤ ਹੈ 16 ਦੀ ਵਰਤੋਂ ਕਰਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਘਾਤਕ ਬੈਕਆਫ. 16ਵੀਂ ਅਸਫਲ ਕੋਸ਼ਿਸ਼ ਤੋਂ ਬਾਅਦ, ਕਾਰਜ ਨੂੰ ਇਸ ਤਰ੍ਹਾਂ ਚਿੰਨ੍ਹਿਤ ਕੀਤਾ ਜਾਵੇਗਾ ਫੇਲ੍ਹ ਹੈ ਅਤੇ ਇਸਨੂੰ ਕਾਫਕਾ ਕਨੈਕਟ REST ਇੰਟਰਫੇਸ ਦੁਆਰਾ ਮੈਨੂਅਲੀ ਰੀਸਟਾਰਟ ਕਰਨ ਦੀ ਲੋੜ ਹੋਵੇਗੀ।
   • ਦੇ ਮਾਮਲੇ ਵਿਚ PostgreSQL ਡਾਟਾ ਖਤਮ ਨਹੀਂ ਹੋਵੇਗਾ, ਕਿਉਂਕਿ ਰਿਪਲੀਕੇਸ਼ਨ ਸਲਾਟ ਦੀ ਵਰਤੋਂ ਨਾਲ ਕਨੈਕਟਰ ਦੁਆਰਾ ਪੜ੍ਹੀਆਂ ਨਹੀਂ ਗਈਆਂ WAL ਫਾਈਲਾਂ ਨੂੰ ਮਿਟਾਉਣ ਤੋਂ ਰੋਕਿਆ ਜਾਵੇਗਾ। ਇਸ ਸਥਿਤੀ ਵਿੱਚ, ਇੱਕ ਨਨੁਕਸਾਨ ਹੈ: ਜੇ ਕਨੈਕਟਰ ਅਤੇ ਡੀਬੀਐਮਐਸ ਵਿਚਕਾਰ ਨੈਟਵਰਕ ਕਨੈਕਟੀਵਿਟੀ ਲੰਬੇ ਸਮੇਂ ਲਈ ਵਿਘਨ ਪਾਉਂਦੀ ਹੈ, ਤਾਂ ਇੱਕ ਸੰਭਾਵਨਾ ਹੈ ਕਿ ਡਿਸਕ ਸਪੇਸ ਖਤਮ ਹੋ ਜਾਵੇਗੀ, ਅਤੇ ਇਸ ਨਾਲ ਪੂਰੇ ਡੀਬੀਐਮਐਸ ਦੀ ਅਸਫਲਤਾ ਹੋ ਸਕਦੀ ਹੈ।
   • ਦੇ ਮਾਮਲੇ ਵਿਚ MySQL ਕਨੈਕਟੀਵਿਟੀ ਰੀਸਟੋਰ ਹੋਣ ਤੋਂ ਪਹਿਲਾਂ ਬਿਨਲੌਗ ਫਾਈਲਾਂ ਨੂੰ ਖੁਦ DBMS ਦੁਆਰਾ ਘੁੰਮਾਇਆ ਜਾ ਸਕਦਾ ਹੈ। ਇਹ ਕਨੈਕਟਰ ਨੂੰ ਅਸਫਲ ਸਥਿਤੀ ਵਿੱਚ ਜਾਣ ਦਾ ਕਾਰਨ ਬਣੇਗਾ, ਅਤੇ ਇਸਨੂੰ ਆਮ ਕਾਰਵਾਈ ਨੂੰ ਬਹਾਲ ਕਰਨ ਲਈ ਬਿਨਲੌਗਸ ਤੋਂ ਪੜ੍ਹਨਾ ਜਾਰੀ ਰੱਖਣ ਲਈ ਸ਼ੁਰੂਆਤੀ ਸਨੈਪਸ਼ਾਟ ਮੋਡ ਵਿੱਚ ਮੁੜ ਚਾਲੂ ਕਰਨ ਦੀ ਲੋੜ ਹੋਵੇਗੀ।
   • 'ਤੇ MongoDB. ਦਸਤਾਵੇਜ਼ਾਂ ਵਿੱਚ ਕਿਹਾ ਗਿਆ ਹੈ: ਲੌਗ/ਓਪਲੌਗ ਫਾਈਲਾਂ ਨੂੰ ਮਿਟਾ ਦਿੱਤੇ ਜਾਣ ਦੀ ਸਥਿਤੀ ਵਿੱਚ ਕਨੈਕਟਰ ਦਾ ਵਿਵਹਾਰ ਅਤੇ ਕਨੈਕਟਰ ਉਸ ਸਥਿਤੀ ਤੋਂ ਪੜ੍ਹਨਾ ਜਾਰੀ ਨਹੀਂ ਰੱਖ ਸਕਦਾ ਜਿੱਥੇ ਇਸਨੂੰ ਛੱਡਿਆ ਗਿਆ ਸੀ, ਸਾਰੇ DBMS ਲਈ ਇੱਕੋ ਜਿਹਾ ਹੈ। ਇਹ ਇਸ ਤੱਥ ਵਿੱਚ ਹੈ ਕਿ ਕਨੈਕਟਰ ਰਾਜ ਵਿੱਚ ਜਾਵੇਗਾ ਫੇਲ੍ਹ ਹੈ ਅਤੇ ਮੋਡ ਵਿੱਚ ਮੁੜ ਚਾਲੂ ਕਰਨ ਦੀ ਲੋੜ ਹੋਵੇਗੀ ਸ਼ੁਰੂਆਤੀ ਸਨੈਪਸ਼ਾਟ.

     ਹਾਲਾਂਕਿ, ਅਪਵਾਦ ਹਨ. ਜੇ ਕਨੈਕਟਰ ਲੰਬੇ ਸਮੇਂ ਲਈ ਡਿਸਕਨੈਕਟ ਕੀਤੀ ਸਥਿਤੀ ਵਿੱਚ ਸੀ (ਜਾਂ ਮੋਂਗੋਡੀਬੀ ਉਦਾਹਰਨ ਤੱਕ ਨਹੀਂ ਪਹੁੰਚ ਸਕਿਆ), ਅਤੇ ਓਪਲੌਗ ਨੂੰ ਇਸ ਸਮੇਂ ਦੌਰਾਨ ਘੁੰਮਾਇਆ ਗਿਆ ਸੀ, ਤਾਂ ਜਦੋਂ ਕਨੈਕਸ਼ਨ ਰੀਸਟੋਰ ਕੀਤਾ ਜਾਂਦਾ ਹੈ, ਤਾਂ ਕਨੈਕਟਰ ਸ਼ਾਂਤ ਰੂਪ ਵਿੱਚ ਪਹਿਲੀ ਉਪਲਬਧ ਸਥਿਤੀ ਤੋਂ ਡੇਟਾ ਨੂੰ ਪੜ੍ਹਨਾ ਜਾਰੀ ਰੱਖੇਗਾ। , ਇਸੇ ਕਰਕੇ ਕਾਫਕਾ ਵਿੱਚ ਕੁਝ ਡੇਟਾ ਨਾ ਮਾਰਿਆ ਜਾਵੇਗਾ.

ਸਿੱਟਾ

ਡੀਬੇਜ਼ੀਅਮ CDC ਪ੍ਰਣਾਲੀਆਂ ਨਾਲ ਮੇਰਾ ਪਹਿਲਾ ਅਨੁਭਵ ਹੈ ਅਤੇ ਸਮੁੱਚੇ ਤੌਰ 'ਤੇ ਬਹੁਤ ਸਕਾਰਾਤਮਕ ਰਿਹਾ ਹੈ। ਪ੍ਰੋਜੈਕਟ ਨੇ ਮੁੱਖ DBMS ਦੇ ਸਮਰਥਨ, ਸੰਰਚਨਾ ਦੀ ਸੌਖ, ਕਲੱਸਟਰਿੰਗ ਲਈ ਸਮਰਥਨ ਅਤੇ ਇੱਕ ਸਰਗਰਮ ਭਾਈਚਾਰੇ ਨੂੰ ਰਿਸ਼ਵਤ ਦਿੱਤੀ। ਅਭਿਆਸ ਵਿੱਚ ਦਿਲਚਸਪੀ ਰੱਖਣ ਵਾਲਿਆਂ ਲਈ, ਮੈਂ ਸਿਫਾਰਸ਼ ਕਰਦਾ ਹਾਂ ਕਿ ਤੁਸੀਂ ਇਸ ਲਈ ਗਾਈਡਾਂ ਨੂੰ ਪੜ੍ਹੋ ਕਾਫਕਾ ਕਨੈਕਟ и ਡੀਬੇਜ਼ੀਅਮ.

ਕਾਫਕਾ ਕਨੈਕਟ ਲਈ JDBC ਕਨੈਕਟਰ ਦੀ ਤੁਲਨਾ ਵਿੱਚ, Debezium ਦਾ ਮੁੱਖ ਫਾਇਦਾ ਇਹ ਹੈ ਕਿ ਤਬਦੀਲੀਆਂ ਨੂੰ DBMS ਲੌਗਸ ਤੋਂ ਪੜ੍ਹਿਆ ਜਾਂਦਾ ਹੈ, ਜੋ ਕਿ ਘੱਟੋ-ਘੱਟ ਦੇਰੀ ਨਾਲ ਡਾਟਾ ਪ੍ਰਾਪਤ ਕਰਨ ਦੀ ਇਜਾਜ਼ਤ ਦਿੰਦਾ ਹੈ। JDBC ਕਨੈਕਟਰ (ਕਾਫਕਾ ਕਨੈਕਟ ਦੁਆਰਾ ਪ੍ਰਦਾਨ ਕੀਤਾ ਗਿਆ) ਇੱਕ ਨਿਸ਼ਚਿਤ ਅੰਤਰਾਲ 'ਤੇ ਟ੍ਰੈਕ ਕੀਤੀ ਸਾਰਣੀ ਤੋਂ ਪੁੱਛਗਿੱਛ ਕਰਦਾ ਹੈ ਅਤੇ (ਇਸੇ ਕਾਰਨ ਕਰਕੇ) ਡੇਟਾ ਨੂੰ ਮਿਟਾਏ ਜਾਣ 'ਤੇ ਸੁਨੇਹੇ ਨਹੀਂ ਬਣਾਉਂਦਾ (ਤੁਸੀਂ ਉਸ ਡੇਟਾ ਲਈ ਪੁੱਛਗਿੱਛ ਕਿਵੇਂ ਕਰ ਸਕਦੇ ਹੋ ਜੋ ਉੱਥੇ ਨਹੀਂ ਹੈ?)।

ਸਮਾਨ ਸਮੱਸਿਆਵਾਂ ਨੂੰ ਹੱਲ ਕਰਨ ਲਈ, ਤੁਸੀਂ ਹੇਠਾਂ ਦਿੱਤੇ ਹੱਲਾਂ (ਡੇਬੇਜ਼ੀਅਮ ਤੋਂ ਇਲਾਵਾ) ਵੱਲ ਧਿਆਨ ਦੇ ਸਕਦੇ ਹੋ:

• ਜੇਡੀਬੀਸੀ ਕਨੈਕਟਰ ਕਾਫਕਾ ਕਨੈਕਟ
• ਕੁਝ MySQL-ਸਿਰਫ ਹੱਲ:
  • MySQL ਸਟ੍ਰੀਮਰ;
  • ਮੈਕਸਵੈੱਲ;
  • ਸਪਾਈਨਲ ਟੈਪ;
• ਓਰੇਕਲ ਗੋਲਡਨ ਗੇਟ, ਪਰ ਇਹ ਇੱਕ ਪੂਰੀ ਤਰ੍ਹਾਂ ਵੱਖਰੀ "ਵਜ਼ਨ ਸ਼੍ਰੇਣੀ" ਹੈ।

PS

ਸਾਡੇ ਬਲੌਗ 'ਤੇ ਵੀ ਪੜ੍ਹੋ:

• «ਕੁਬਰਨੇਟਸ ਵਿੱਚ ਕਾਫਕਾ ਕਲੱਸਟਰ ਲਈ ਉਚਿਤ ਆਕਾਰ ਨਿਰਧਾਰਤ ਕਰੋ";
• «ਸਾਡੇ SRE ਰੋਜ਼ਾਨਾ ਜੀਵਨ ਤੋਂ ਵਿਹਾਰਕ ਕਹਾਣੀਆਂ। ਭਾਗ 2";
• «ਕੁਬਰਨੇਟਸ, ਸਾਡੀਆਂ ਚੋਣਾਂ ਅਤੇ ਅਨੁਭਵ ਲਈ PostgreSQL ਸਟੇਟਮੈਂਟਾਂ ਦੀ ਇੱਕ ਸੰਖੇਪ ਜਾਣਕਾਰੀ".

ਸਰੋਤ: www.habr.com