ਅਪਾਚੇ ਕਾਫਕਾ ਅਤੇ ਸਪਾਰਕ ਸਟ੍ਰੀਮਿੰਗ ਦੇ ਨਾਲ ਸਟ੍ਰੀਮਿੰਗ ਡੇਟਾ ਪ੍ਰੋਸੈਸਿੰਗ

ਹੈਲੋ, ਹੈਬਰ! ਅੱਜ ਅਸੀਂ ਇੱਕ ਸਿਸਟਮ ਬਣਾਵਾਂਗੇ ਜੋ ਸਪਾਰਕ ਸਟ੍ਰੀਮਿੰਗ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ ਅਪਾਚੇ ਕਾਫਕਾ ਸੰਦੇਸ਼ ਸਟ੍ਰੀਮਾਂ ਦੀ ਪ੍ਰਕਿਰਿਆ ਕਰੇਗਾ ਅਤੇ ਪ੍ਰੋਸੈਸਿੰਗ ਨਤੀਜਿਆਂ ਨੂੰ AWS RDS ਕਲਾਉਡ ਡੇਟਾਬੇਸ ਵਿੱਚ ਲਿਖੇਗਾ।

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

ਜਾਦੂ ਅਤੇ ਜਾਦੂ ਦੇ ਜਾਦੂ ਦੀ ਵਰਤੋਂ ਕੀਤੇ ਬਿਨਾਂ ਇਸ ਕੇਸ ਨੂੰ ਕਿਵੇਂ ਲਾਗੂ ਕਰਨਾ ਹੈ - ਕੱਟ ਦੇ ਹੇਠਾਂ ਪੜ੍ਹੋ! ਜਾਣਾ!

(ਚਿੱਤਰ ਸਰੋਤ)

ਜਾਣ ਪਛਾਣ

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

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

ਵਿਕਸਤ ਸਿਸਟਮ ਦਾ ਆਰਕੀਟੈਕਚਰ

ਵਰਤੇ ਗਏ ਹਿੱਸੇ:

• ਅਪਾਚੇ ਕਾਫਕਾ ਇੱਕ ਵੰਡਿਆ ਪਬਲਿਸ਼-ਸਬਸਕ੍ਰਾਈਬ ਮੈਸੇਜਿੰਗ ਸਿਸਟਮ ਹੈ। ਔਫਲਾਈਨ ਅਤੇ ਔਨਲਾਈਨ ਸੁਨੇਹੇ ਦੀ ਖਪਤ ਲਈ ਢੁਕਵਾਂ। ਡੇਟਾ ਦੇ ਨੁਕਸਾਨ ਨੂੰ ਰੋਕਣ ਲਈ, ਕਾਫਕਾ ਸੁਨੇਹਿਆਂ ਨੂੰ ਡਿਸਕ 'ਤੇ ਸਟੋਰ ਕੀਤਾ ਜਾਂਦਾ ਹੈ ਅਤੇ ਕਲੱਸਟਰ ਦੇ ਅੰਦਰ ਦੁਹਰਾਇਆ ਜਾਂਦਾ ਹੈ। ਕਾਫਕਾ ਸਿਸਟਮ ਜ਼ੂਕੀਪਰ ਸਿੰਕ੍ਰੋਨਾਈਜ਼ੇਸ਼ਨ ਸੇਵਾ ਦੇ ਸਿਖਰ 'ਤੇ ਬਣਾਇਆ ਗਿਆ ਹੈ;
• ਅਪਾਚੇ ਸਪਾਰਕ ਸਟ੍ਰੀਮਿੰਗ - ਸਟ੍ਰੀਮਿੰਗ ਡੇਟਾ ਦੀ ਪ੍ਰਕਿਰਿਆ ਲਈ ਸਪਾਰਕ ਕੰਪੋਨੈਂਟ। ਸਪਾਰਕ ਸਟ੍ਰੀਮਿੰਗ ਮੋਡੀਊਲ ਇੱਕ ਮਾਈਕ੍ਰੋ-ਬੈਚ ਆਰਕੀਟੈਕਚਰ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਬਣਾਇਆ ਗਿਆ ਹੈ, ਜਿੱਥੇ ਡੇਟਾ ਸਟ੍ਰੀਮ ਨੂੰ ਛੋਟੇ ਡੇਟਾ ਪੈਕੇਟਾਂ ਦੇ ਇੱਕ ਨਿਰੰਤਰ ਕ੍ਰਮ ਵਜੋਂ ਵਿਆਖਿਆ ਕੀਤੀ ਜਾਂਦੀ ਹੈ। ਸਪਾਰਕ ਸਟ੍ਰੀਮਿੰਗ ਵੱਖ-ਵੱਖ ਸਰੋਤਾਂ ਤੋਂ ਡੇਟਾ ਲੈਂਦਾ ਹੈ ਅਤੇ ਇਸਨੂੰ ਛੋਟੇ ਪੈਕੇਜਾਂ ਵਿੱਚ ਜੋੜਦਾ ਹੈ। ਨਵੇਂ ਪੈਕੇਜ ਨਿਯਮਤ ਅੰਤਰਾਲਾਂ 'ਤੇ ਬਣਾਏ ਜਾਂਦੇ ਹਨ। ਹਰ ਵਾਰ ਅੰਤਰਾਲ ਦੀ ਸ਼ੁਰੂਆਤ ਵਿੱਚ, ਇੱਕ ਨਵਾਂ ਪੈਕੇਟ ਬਣਾਇਆ ਜਾਂਦਾ ਹੈ, ਅਤੇ ਉਸ ਅੰਤਰਾਲ ਦੌਰਾਨ ਪ੍ਰਾਪਤ ਕੀਤਾ ਕੋਈ ਵੀ ਡੇਟਾ ਪੈਕੇਟ ਵਿੱਚ ਸ਼ਾਮਲ ਕੀਤਾ ਜਾਂਦਾ ਹੈ। ਅੰਤਰਾਲ ਦੇ ਅੰਤ 'ਤੇ, ਪੈਕੇਟ ਦਾ ਵਾਧਾ ਰੁਕ ਜਾਂਦਾ ਹੈ. ਅੰਤਰਾਲ ਦਾ ਆਕਾਰ ਇੱਕ ਪੈਰਾਮੀਟਰ ਦੁਆਰਾ ਨਿਰਧਾਰਤ ਕੀਤਾ ਜਾਂਦਾ ਹੈ ਜਿਸਨੂੰ ਬੈਚ ਅੰਤਰਾਲ ਕਿਹਾ ਜਾਂਦਾ ਹੈ;
• ਅਪਾਚੇ ਸਪਾਰਕ SQL - ਸਪਾਰਕ ਫੰਕਸ਼ਨਲ ਪ੍ਰੋਗਰਾਮਿੰਗ ਨਾਲ ਰਿਲੇਸ਼ਨਲ ਪ੍ਰੋਸੈਸਿੰਗ ਨੂੰ ਜੋੜਦਾ ਹੈ। ਸਟ੍ਰਕਚਰਡ ਡੇਟਾ ਦਾ ਅਰਥ ਹੈ ਉਹ ਡੇਟਾ ਜਿਸਦਾ ਇੱਕ ਸਕੀਮਾ ਹੈ, ਭਾਵ, ਸਾਰੇ ਰਿਕਾਰਡਾਂ ਲਈ ਖੇਤਰਾਂ ਦਾ ਇੱਕ ਸਿੰਗਲ ਸੈੱਟ। ਸਪਾਰਕ SQL ਕਈ ਤਰ੍ਹਾਂ ਦੇ ਢਾਂਚਾਗਤ ਡੇਟਾ ਸਰੋਤਾਂ ਤੋਂ ਇਨਪੁਟ ਦਾ ਸਮਰਥਨ ਕਰਦਾ ਹੈ ਅਤੇ, ਸਕੀਮਾ ਜਾਣਕਾਰੀ ਦੀ ਉਪਲਬਧਤਾ ਲਈ ਧੰਨਵਾਦ, ਇਹ ਕੁਸ਼ਲਤਾ ਨਾਲ ਰਿਕਾਰਡਾਂ ਦੇ ਸਿਰਫ ਲੋੜੀਂਦੇ ਖੇਤਰਾਂ ਨੂੰ ਪ੍ਰਾਪਤ ਕਰ ਸਕਦਾ ਹੈ, ਅਤੇ ਡਾਟਾਫ੍ਰੇਮ APIs ਵੀ ਪ੍ਰਦਾਨ ਕਰਦਾ ਹੈ;
• AWS RDS ਇੱਕ ਮੁਕਾਬਲਤਨ ਸਸਤਾ ਕਲਾਉਡ-ਅਧਾਰਿਤ ਰਿਲੇਸ਼ਨਲ ਡੇਟਾਬੇਸ, ਵੈੱਬ ਸੇਵਾ ਹੈ ਜੋ ਸੈੱਟਅੱਪ, ਸੰਚਾਲਨ ਅਤੇ ਸਕੇਲਿੰਗ ਨੂੰ ਸਰਲ ਬਣਾਉਂਦਾ ਹੈ, ਅਤੇ ਸਿੱਧੇ ਐਮਾਜ਼ਾਨ ਦੁਆਰਾ ਪ੍ਰਬੰਧਿਤ ਕੀਤਾ ਜਾਂਦਾ ਹੈ।

ਕਾਫਕਾ ਸਰਵਰ ਨੂੰ ਸਥਾਪਿਤ ਅਤੇ ਚਲਾਉਣਾ

ਕਾਫਕਾ ਦੀ ਸਿੱਧੀ ਵਰਤੋਂ ਕਰਨ ਤੋਂ ਪਹਿਲਾਂ, ਤੁਹਾਨੂੰ ਇਹ ਯਕੀਨੀ ਬਣਾਉਣ ਦੀ ਲੋੜ ਹੈ ਕਿ ਤੁਹਾਡੇ ਕੋਲ ਜਾਵਾ ਹੈ, ਕਿਉਂਕਿ... JVM ਕੰਮ ਲਈ ਵਰਤਿਆ ਜਾਂਦਾ ਹੈ:

sudo apt-get update 
sudo apt-get install default-jre
java -version

ਕਾਫਕਾ ਦੇ ਨਾਲ ਕੰਮ ਕਰਨ ਲਈ ਇੱਕ ਨਵਾਂ ਉਪਭੋਗਤਾ ਬਣਾਓ:

sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo

ਅੱਗੇ, ਅਧਿਕਾਰਤ ਅਪਾਚੇ ਕਾਫਕਾ ਵੈਬਸਾਈਟ ਤੋਂ ਵੰਡ ਨੂੰ ਡਾਉਨਲੋਡ ਕਰੋ:

wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"

ਡਾਊਨਲੋਡ ਕੀਤੇ ਆਰਕਾਈਵ ਨੂੰ ਅਨਪੈਕ ਕਰੋ:

tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka

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

vim ~/kafka/config/server.properties

ਫਾਈਲ ਦੇ ਅੰਤ ਵਿੱਚ ਹੇਠਾਂ ਦਿੱਤੇ ਨੂੰ ਸ਼ਾਮਲ ਕਰੋ:

delete.topic.enable = true

ਕਾਫਕਾ ਸਰਵਰ ਸ਼ੁਰੂ ਕਰਨ ਤੋਂ ਪਹਿਲਾਂ, ਤੁਹਾਨੂੰ ਜ਼ੂਕੀਪਰ ਸਰਵਰ ਸ਼ੁਰੂ ਕਰਨ ਦੀ ਲੋੜ ਹੈ; ਅਸੀਂ ਸਹਾਇਕ ਸਕ੍ਰਿਪਟ ਦੀ ਵਰਤੋਂ ਕਰਾਂਗੇ ਜੋ ਕਾਫਕਾ ਵੰਡ ਦੇ ਨਾਲ ਆਉਂਦੀ ਹੈ:

Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties

ZooKeeper ਦੇ ਸਫਲਤਾਪੂਰਵਕ ਸ਼ੁਰੂ ਹੋਣ ਤੋਂ ਬਾਅਦ, ਕਾਫਕਾ ਸਰਵਰ ਨੂੰ ਇੱਕ ਵੱਖਰੇ ਟਰਮੀਨਲ ਵਿੱਚ ਲਾਂਚ ਕਰੋ:

bin/kafka-server-start.sh config/server.properties

ਆਉ ਟ੍ਰਾਂਜੈਕਸ਼ਨ ਨਾਮਕ ਇੱਕ ਨਵਾਂ ਵਿਸ਼ਾ ਬਣਾਈਏ:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transaction

ਚਲੋ ਇਹ ਸੁਨਿਸ਼ਚਿਤ ਕਰੀਏ ਕਿ ਭਾਗਾਂ ਅਤੇ ਪ੍ਰਤੀਕ੍ਰਿਤੀ ਦੀ ਲੋੜੀਂਦੀ ਗਿਣਤੀ ਵਾਲਾ ਵਿਸ਼ਾ ਬਣਾਇਆ ਗਿਆ ਹੈ:

bin/kafka-topics.sh --describe --zookeeper localhost:2181

ਆਉ ਨਵੇਂ ਬਣੇ ਵਿਸ਼ੇ ਲਈ ਉਤਪਾਦਕ ਅਤੇ ਖਪਤਕਾਰ ਦੀ ਜਾਂਚ ਕਰਨ ਦੇ ਪਲਾਂ ਨੂੰ ਗੁਆ ਦੇਈਏ. ਤੁਸੀਂ ਸੁਨੇਹੇ ਭੇਜਣ ਅਤੇ ਪ੍ਰਾਪਤ ਕਰਨ ਦੀ ਜਾਂਚ ਕਿਵੇਂ ਕਰ ਸਕਦੇ ਹੋ ਇਸ ਬਾਰੇ ਹੋਰ ਵੇਰਵੇ ਅਧਿਕਾਰਤ ਦਸਤਾਵੇਜ਼ਾਂ ਵਿੱਚ ਲਿਖੇ ਗਏ ਹਨ - ਕੁਝ ਸੁਨੇਹੇ ਭੇਜੋ. ਖੈਰ, ਅਸੀਂ KafkaProducer API ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ ਪਾਈਥਨ ਵਿੱਚ ਇੱਕ ਨਿਰਮਾਤਾ ਲਿਖਣ ਲਈ ਅੱਗੇ ਵਧਦੇ ਹਾਂ।

ਨਿਰਮਾਤਾ ਲਿਖਣਾ

ਨਿਰਮਾਤਾ ਬੇਤਰਤੀਬ ਡੇਟਾ ਤਿਆਰ ਕਰੇਗਾ - ਹਰ ਸਕਿੰਟ ਵਿੱਚ 100 ਸੁਨੇਹੇ। ਬੇਤਰਤੀਬ ਡੇਟਾ ਦੁਆਰਾ ਸਾਡਾ ਮਤਲਬ ਇੱਕ ਸ਼ਬਦਕੋਸ਼ ਹੈ ਜਿਸ ਵਿੱਚ ਤਿੰਨ ਖੇਤਰਾਂ ਸ਼ਾਮਲ ਹਨ:

• ਸ਼ਾਖਾ - ਕ੍ਰੈਡਿਟ ਸੰਸਥਾ ਦੇ ਵਿਕਰੀ ਸਥਾਨ ਦਾ ਨਾਮ;
• ਕਰੰਸੀ - ਲੈਣ-ਦੇਣ ਦੀ ਮੁਦਰਾ;
• ਮਾਤਰਾ - ਲੈਣ-ਦੇਣ ਦੀ ਰਕਮ। ਜੇਕਰ ਇਹ ਬੈਂਕ ਦੁਆਰਾ ਮੁਦਰਾ ਦੀ ਖਰੀਦ ਹੈ ਤਾਂ ਰਕਮ ਇੱਕ ਸਕਾਰਾਤਮਕ ਸੰਖਿਆ ਹੋਵੇਗੀ, ਅਤੇ ਜੇਕਰ ਇਹ ਇੱਕ ਵਿਕਰੀ ਹੈ ਤਾਂ ਇੱਕ ਨਕਾਰਾਤਮਕ ਸੰਖਿਆ ਹੋਵੇਗੀ।

ਨਿਰਮਾਤਾ ਲਈ ਕੋਡ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਦਾ ਹੈ:

from numpy.random import choice, randint

def get_random_value():
    new_dict = {}

    branch_list = ['Kazan', 'SPB', 'Novosibirsk', 'Surgut']
    currency_list = ['RUB', 'USD', 'EUR', 'GBP']

    new_dict['branch'] = choice(branch_list)
    new_dict['currency'] = choice(currency_list)
    new_dict['amount'] = randint(-100, 100)

    return new_dict

ਅੱਗੇ, ਭੇਜਣ ਵਿਧੀ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ, ਅਸੀਂ ਸਰਵਰ ਨੂੰ ਇੱਕ ਸੁਨੇਹਾ ਭੇਜਦੇ ਹਾਂ, ਜਿਸ ਵਿਸ਼ੇ ਦੀ ਸਾਨੂੰ ਲੋੜ ਹੈ, JSON ਫਾਰਮੈਟ ਵਿੱਚ:

from kafka import KafkaProducer    

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                             value_serializer=lambda x:dumps(x).encode('utf-8'),
                             compression_type='gzip')
my_topic = 'transaction'
data = get_random_value()

try:
    future = producer.send(topic = my_topic, value = data)
    record_metadata = future.get(timeout=10)
    
    print('--> The message has been sent to a topic: 
            {}, partition: {}, offset: {}' 
            .format(record_metadata.topic,
                record_metadata.partition,
                record_metadata.offset ))   
                             
except Exception as e:
    print('--> It seems an Error occurred: {}'.format(e))

finally:
    producer.flush()

ਸਕ੍ਰਿਪਟ ਚਲਾਉਂਦੇ ਸਮੇਂ, ਸਾਨੂੰ ਟਰਮੀਨਲ ਵਿੱਚ ਹੇਠਾਂ ਦਿੱਤੇ ਸੁਨੇਹੇ ਪ੍ਰਾਪਤ ਹੁੰਦੇ ਹਨ:

ਇਸਦਾ ਮਤਲਬ ਹੈ ਕਿ ਸਭ ਕੁਝ ਉਸੇ ਤਰ੍ਹਾਂ ਕੰਮ ਕਰਦਾ ਹੈ ਜਿਵੇਂ ਅਸੀਂ ਚਾਹੁੰਦੇ ਹਾਂ - ਨਿਰਮਾਤਾ ਸਾਨੂੰ ਲੋੜੀਂਦੇ ਵਿਸ਼ੇ 'ਤੇ ਸੰਦੇਸ਼ ਬਣਾਉਂਦਾ ਹੈ ਅਤੇ ਭੇਜਦਾ ਹੈ।
ਅਗਲਾ ਕਦਮ ਸਪਾਰਕ ਨੂੰ ਸਥਾਪਿਤ ਕਰਨਾ ਅਤੇ ਇਸ ਸੰਦੇਸ਼ ਸਟ੍ਰੀਮ ਦੀ ਪ੍ਰਕਿਰਿਆ ਕਰਨਾ ਹੈ।

ਅਪਾਚੇ ਸਪਾਰਕ ਸਥਾਪਤ ਕਰਨਾ

ਅਪਾਚੇ ਸਪਾਰਕ ਇੱਕ ਯੂਨੀਵਰਸਲ ਅਤੇ ਉੱਚ-ਪ੍ਰਦਰਸ਼ਨ ਵਾਲਾ ਕਲੱਸਟਰ ਕੰਪਿਊਟਿੰਗ ਪਲੇਟਫਾਰਮ ਹੈ।

ਸਪਾਰਕ ਇੰਟਰਐਕਟਿਵ ਪੁੱਛਗਿੱਛਾਂ ਅਤੇ ਸਟ੍ਰੀਮ ਪ੍ਰੋਸੈਸਿੰਗ ਸਮੇਤ ਗਣਨਾ ਕਿਸਮਾਂ ਦੀ ਇੱਕ ਵਿਸ਼ਾਲ ਸ਼੍ਰੇਣੀ ਦਾ ਸਮਰਥਨ ਕਰਦੇ ਹੋਏ MapReduce ਮਾਡਲ ਦੇ ਪ੍ਰਸਿੱਧ ਲਾਗੂਕਰਨ ਨਾਲੋਂ ਬਿਹਤਰ ਪ੍ਰਦਰਸ਼ਨ ਕਰਦਾ ਹੈ। ਵੱਡੀ ਮਾਤਰਾ ਵਿੱਚ ਡੇਟਾ ਦੀ ਪ੍ਰਕਿਰਿਆ ਕਰਨ ਵੇਲੇ ਸਪੀਡ ਇੱਕ ਮਹੱਤਵਪੂਰਣ ਭੂਮਿਕਾ ਨਿਭਾਉਂਦੀ ਹੈ, ਕਿਉਂਕਿ ਇਹ ਉਹ ਗਤੀ ਹੈ ਜੋ ਤੁਹਾਨੂੰ ਮਿੰਟਾਂ ਜਾਂ ਘੰਟਿਆਂ ਦੀ ਉਡੀਕ ਕੀਤੇ ਬਿਨਾਂ ਇੰਟਰਐਕਟਿਵ ਤਰੀਕੇ ਨਾਲ ਕੰਮ ਕਰਨ ਦੀ ਆਗਿਆ ਦਿੰਦੀ ਹੈ। ਸਪਾਰਕ ਦੀ ਸਭ ਤੋਂ ਵੱਡੀ ਤਾਕਤ ਜੋ ਇਸਨੂੰ ਇੰਨੀ ਤੇਜ਼ੀ ਨਾਲ ਬਣਾਉਂਦੀ ਹੈ, ਇਹ ਇਨ-ਮੈਮੋਰੀ ਗਣਨਾ ਕਰਨ ਦੀ ਯੋਗਤਾ ਹੈ।

ਇਹ ਫਰੇਮਵਰਕ Scala ਵਿੱਚ ਲਿਖਿਆ ਗਿਆ ਹੈ, ਇਸ ਲਈ ਤੁਹਾਨੂੰ ਇਸਨੂੰ ਪਹਿਲਾਂ ਸਥਾਪਿਤ ਕਰਨ ਦੀ ਲੋੜ ਹੈ:

sudo apt-get install scala

ਅਧਿਕਾਰਤ ਵੈੱਬਸਾਈਟ ਤੋਂ ਸਪਾਰਕ ਵੰਡ ਨੂੰ ਡਾਊਨਲੋਡ ਕਰੋ:

wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"

ਆਰਕਾਈਵ ਨੂੰ ਅਨਪੈਕ ਕਰੋ:

sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/spark

ਬੈਸ਼ ਫਾਈਲ ਵਿੱਚ ਸਪਾਰਕ ਦਾ ਮਾਰਗ ਜੋੜੋ:

vim ~/.bashrc

ਸੰਪਾਦਕ ਦੁਆਰਾ ਹੇਠ ਲਿਖੀਆਂ ਲਾਈਨਾਂ ਜੋੜੋ:

SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH

bashrc ਵਿੱਚ ਬਦਲਾਅ ਕਰਨ ਤੋਂ ਬਾਅਦ ਹੇਠਾਂ ਦਿੱਤੀ ਕਮਾਂਡ ਚਲਾਓ:

source ~/.bashrc

AWS PostgreSQL ਨੂੰ ਤੈਨਾਤ ਕੀਤਾ ਜਾ ਰਿਹਾ ਹੈ

ਜੋ ਬਚਿਆ ਹੈ ਉਹ ਡੇਟਾਬੇਸ ਨੂੰ ਤੈਨਾਤ ਕਰਨਾ ਹੈ ਜਿਸ ਵਿੱਚ ਅਸੀਂ ਸਟ੍ਰੀਮਾਂ ਤੋਂ ਪ੍ਰਕਿਰਿਆ ਕੀਤੀ ਜਾਣਕਾਰੀ ਨੂੰ ਅਪਲੋਡ ਕਰਾਂਗੇ। ਇਸਦੇ ਲਈ ਅਸੀਂ AWS RDS ਸੇਵਾ ਦੀ ਵਰਤੋਂ ਕਰਾਂਗੇ।

AWS ਕੰਸੋਲ -> AWS RDS -> ਡਾਟਾਬੇਸ -> ਡਾਟਾਬੇਸ ਬਣਾਓ 'ਤੇ ਜਾਓ:

PostgreSQL ਚੁਣੋ ਅਤੇ ਅੱਗੇ ਕਲਿੱਕ ਕਰੋ:

ਕਿਉਂਕਿ ਇਹ ਉਦਾਹਰਨ ਸਿਰਫ਼ ਵਿਦਿਅਕ ਉਦੇਸ਼ਾਂ ਲਈ ਹੈ; ਅਸੀਂ "ਘੱਟੋ-ਘੱਟ" (ਮੁਫ਼ਤ ਟੀਅਰ) ਇੱਕ ਮੁਫ਼ਤ ਸਰਵਰ ਦੀ ਵਰਤੋਂ ਕਰਾਂਗੇ:

ਅੱਗੇ, ਅਸੀਂ ਫ੍ਰੀ ਟੀਅਰ ਬਲਾਕ ਵਿੱਚ ਇੱਕ ਟਿਕ ਲਗਾ ਦਿੰਦੇ ਹਾਂ, ਅਤੇ ਉਸ ਤੋਂ ਬਾਅਦ ਸਾਨੂੰ ਆਪਣੇ ਆਪ ਹੀ t2.micro ਕਲਾਸ ਦੀ ਇੱਕ ਉਦਾਹਰਣ ਪੇਸ਼ ਕੀਤੀ ਜਾਵੇਗੀ - ਹਾਲਾਂਕਿ ਇਹ ਕਮਜ਼ੋਰ ਹੈ, ਇਹ ਸਾਡੇ ਕੰਮ ਲਈ ਮੁਫਤ ਅਤੇ ਕਾਫ਼ੀ ਢੁਕਵਾਂ ਹੈ:

ਅੱਗੇ ਬਹੁਤ ਮਹੱਤਵਪੂਰਨ ਚੀਜ਼ਾਂ ਆਉਂਦੀਆਂ ਹਨ: ਡੇਟਾਬੇਸ ਉਦਾਹਰਣ ਦਾ ਨਾਮ, ਮਾਸਟਰ ਉਪਭੋਗਤਾ ਦਾ ਨਾਮ ਅਤੇ ਉਸਦਾ ਪਾਸਵਰਡ। ਆਓ ਉਦਾਹਰਣ ਨੂੰ ਨਾਮ ਦੇਈਏ: myHabrTest, ਮਾਸਟਰ ਉਪਭੋਗਤਾ: habr, ਪਾਸਵਰਡ: habr12345 ਅਤੇ ਅੱਗੇ ਬਟਨ 'ਤੇ ਕਲਿੱਕ ਕਰੋ:

ਅਗਲੇ ਪੰਨੇ 'ਤੇ ਸਾਡੇ ਡੇਟਾਬੇਸ ਸਰਵਰ ਦੀ ਬਾਹਰੋਂ ਪਹੁੰਚਯੋਗਤਾ (ਜਨਤਕ ਪਹੁੰਚਯੋਗਤਾ) ਅਤੇ ਪੋਰਟ ਉਪਲਬਧਤਾ ਲਈ ਜ਼ਿੰਮੇਵਾਰ ਮਾਪਦੰਡ ਹਨ:

ਚਲੋ VPC ਸੁਰੱਖਿਆ ਸਮੂਹ ਲਈ ਇੱਕ ਨਵੀਂ ਸੈਟਿੰਗ ਬਣਾਈਏ, ਜੋ ਪੋਰਟ 5432 (PostgreSQL) ਰਾਹੀਂ ਸਾਡੇ ਡੇਟਾਬੇਸ ਸਰਵਰ ਤੱਕ ਬਾਹਰੀ ਪਹੁੰਚ ਦੀ ਆਗਿਆ ਦੇਵੇਗੀ।
ਚਲੋ ਇੱਕ ਵੱਖਰੀ ਬ੍ਰਾਊਜ਼ਰ ਵਿੰਡੋ ਵਿੱਚ AWS ਕੰਸੋਲ ਤੇ VPC ਡੈਸ਼ਬੋਰਡ -> ਸੁਰੱਖਿਆ ਸਮੂਹ -> ਸੁਰੱਖਿਆ ਸਮੂਹ ਸੈਕਸ਼ਨ ਬਣਾਓ:

ਅਸੀਂ ਸੁਰੱਖਿਆ ਸਮੂਹ ਲਈ ਨਾਮ ਸੈੱਟ ਕੀਤਾ ਹੈ - PostgreSQL, ਇੱਕ ਵੇਰਵਾ, ਇਹ ਦਰਸਾਉਂਦਾ ਹੈ ਕਿ ਇਹ ਸਮੂਹ ਕਿਸ VPC ਨਾਲ ਸੰਬੰਧਿਤ ਹੋਣਾ ਚਾਹੀਦਾ ਹੈ ਅਤੇ ਬਣਾਓ ਬਟਨ 'ਤੇ ਕਲਿੱਕ ਕਰੋ:

ਨਵੇਂ ਬਣਾਏ ਗਏ ਸਮੂਹ ਲਈ ਪੋਰਟ 5432 ਲਈ ਇਨਬਾਉਂਡ ਨਿਯਮਾਂ ਨੂੰ ਭਰੋ, ਜਿਵੇਂ ਕਿ ਹੇਠਾਂ ਤਸਵੀਰ ਵਿੱਚ ਦਿਖਾਇਆ ਗਿਆ ਹੈ। ਤੁਸੀਂ ਪੋਰਟ ਨੂੰ ਹੱਥੀਂ ਨਿਰਧਾਰਿਤ ਨਹੀਂ ਕਰ ਸਕਦੇ ਹੋ, ਪਰ ਟਾਈਪ ਡ੍ਰੌਪ-ਡਾਉਨ ਸੂਚੀ ਵਿੱਚੋਂ PostgreSQL ਦੀ ਚੋਣ ਕਰੋ।

ਸਖਤੀ ਨਾਲ ਬੋਲਦੇ ਹੋਏ, ਮੁੱਲ ::/0 ਦਾ ਅਰਥ ਹੈ ਪੂਰੀ ਦੁਨੀਆ ਤੋਂ ਸਰਵਰ 'ਤੇ ਆਉਣ ਵਾਲੇ ਟ੍ਰੈਫਿਕ ਦੀ ਉਪਲਬਧਤਾ, ਜੋ ਕਿ ਪ੍ਰਮਾਣਿਕ ​​ਤੌਰ 'ਤੇ ਪੂਰੀ ਤਰ੍ਹਾਂ ਸਹੀ ਨਹੀਂ ਹੈ, ਪਰ ਉਦਾਹਰਣ ਦਾ ਵਿਸ਼ਲੇਸ਼ਣ ਕਰਨ ਲਈ, ਆਓ ਆਪਣੇ ਆਪ ਨੂੰ ਇਸ ਪਹੁੰਚ ਦੀ ਵਰਤੋਂ ਕਰਨ ਦੀ ਇਜਾਜ਼ਤ ਦੇਈਏ:

ਅਸੀਂ ਬ੍ਰਾਊਜ਼ਰ ਪੰਨੇ 'ਤੇ ਵਾਪਸ ਆਉਂਦੇ ਹਾਂ, ਜਿੱਥੇ ਸਾਡੇ ਕੋਲ "ਐਡਵਾਂਸਡ ਸੈਟਿੰਗਜ਼ ਕੌਂਫਿਗਰ ਕਰੋ" ਖੁੱਲੀ ਹੈ ਅਤੇ VPC ਸੁਰੱਖਿਆ ਸਮੂਹ ਭਾਗ ਵਿੱਚ ਚੁਣੋ -> ਮੌਜੂਦਾ VPC ਸੁਰੱਖਿਆ ਸਮੂਹ ਚੁਣੋ -> PostgreSQL:

ਅੱਗੇ, ਡੇਟਾਬੇਸ ਵਿਕਲਪਾਂ ਵਿੱਚ -> ਡੇਟਾਬੇਸ ਨਾਮ -> ਨਾਮ ਸੈੱਟ ਕਰੋ - habrDB.

ਅਸੀਂ ਡਿਫੌਲਟ ਤੌਰ 'ਤੇ, ਬੈਕਅੱਪ (ਬੈਕਅੱਪ ਧਾਰਨ ਦੀ ਮਿਆਦ - 0 ਦਿਨ), ਨਿਗਰਾਨੀ ਅਤੇ ਪ੍ਰਦਰਸ਼ਨ ਇਨਸਾਈਟਸ ਨੂੰ ਅਯੋਗ ਕਰਨ ਦੇ ਅਪਵਾਦ ਦੇ ਨਾਲ, ਬਾਕੀ ਮਾਪਦੰਡਾਂ ਨੂੰ ਛੱਡ ਸਕਦੇ ਹਾਂ। ਬਟਨ 'ਤੇ ਕਲਿੱਕ ਕਰੋ ਡਾਟਾਬੇਸ ਬਣਾਓ:

ਥਰਿੱਡ ਹੈਂਡਲਰ

ਅੰਤਮ ਪੜਾਅ ਇੱਕ ਸਪਾਰਕ ਜੌਬ ਦਾ ਵਿਕਾਸ ਹੋਵੇਗਾ, ਜੋ ਹਰ ਦੋ ਸਕਿੰਟਾਂ ਵਿੱਚ ਕਾਫਕਾ ਤੋਂ ਆਉਣ ਵਾਲੇ ਨਵੇਂ ਡੇਟਾ ਦੀ ਪ੍ਰਕਿਰਿਆ ਕਰੇਗਾ ਅਤੇ ਨਤੀਜਾ ਡੇਟਾਬੇਸ ਵਿੱਚ ਦਾਖਲ ਕਰੇਗਾ।

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

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

streamingContext.checkpoint(checkpointDirectory)

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

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

ਅਸੀਂ KafkaUtils ਲਾਇਬ੍ਰੇਰੀ ਦੀ createDirectStream ਵਿਧੀ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ "ਟ੍ਰਾਂਜੈਕਸ਼ਨ" ਵਿਸ਼ੇ ਨਾਲ ਜੁੜਨ ਲਈ ਇੱਕ ਡਾਇਰੈਕਟਸਟ੍ਰੀਮ ਆਬਜੈਕਟ ਬਣਾਉਂਦੇ ਹਾਂ:

from pyspark.streaming.kafka import KafkaUtils
    
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 2)

broker_list = 'localhost:9092'
topic = 'transaction'

directKafkaStream = KafkaUtils.createDirectStream(ssc,
                                [topic],
                                {"metadata.broker.list": broker_list})

JSON ਫਾਰਮੈਟ ਵਿੱਚ ਆਉਣ ਵਾਲੇ ਡੇਟਾ ਨੂੰ ਪਾਰਸ ਕਰਨਾ:

rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
                                       currency=w['currency'],
                                       amount=w['amount']))
                                       
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")

ਸਪਾਰਕ SQL ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ, ਅਸੀਂ ਇੱਕ ਸਧਾਰਨ ਗਰੁੱਪਿੰਗ ਕਰਦੇ ਹਾਂ ਅਤੇ ਨਤੀਜਾ ਕੰਸੋਲ ਵਿੱਚ ਪ੍ਰਦਰਸ਼ਿਤ ਕਰਦੇ ਹਾਂ:

select 
    from_unixtime(unix_timestamp()) as curr_time,
    t.branch                        as branch_name,
    t.currency                      as currency_code,
    sum(amount)                     as batch_value
from treasury_stream t
group by
    t.branch,
    t.currency

ਪੁੱਛਗਿੱਛ ਟੈਕਸਟ ਨੂੰ ਪ੍ਰਾਪਤ ਕਰਨਾ ਅਤੇ ਇਸਨੂੰ ਸਪਾਰਕ SQL ਦੁਆਰਾ ਚਲਾਉਣਾ:

sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)

ਅਤੇ ਫਿਰ ਅਸੀਂ AWS RDS ਵਿੱਚ ਇੱਕ ਸਾਰਣੀ ਵਿੱਚ ਨਤੀਜੇ ਵਜੋਂ ਇਕੱਠੇ ਕੀਤੇ ਡੇਟਾ ਨੂੰ ਸੁਰੱਖਿਅਤ ਕਰਦੇ ਹਾਂ। ਇੱਕ ਡੇਟਾਬੇਸ ਟੇਬਲ ਵਿੱਚ ਏਗਰੀਗੇਸ਼ਨ ਨਤੀਜਿਆਂ ਨੂੰ ਸੁਰੱਖਿਅਤ ਕਰਨ ਲਈ, ਅਸੀਂ ਡੇਟਾਫ੍ਰੇਮ ਆਬਜੈਕਟ ਦੀ ਲਿਖਣ ਵਿਧੀ ਦੀ ਵਰਤੋਂ ਕਰਾਂਗੇ:

testResultDataFrame.write 
    .format("jdbc") 
    .mode("append") 
    .option("driver", 'org.postgresql.Driver') 
    .option("url","jdbc:postgresql://myhabrtest.ciny8bykwxeg.us-east-1.rds.amazonaws.com:5432/habrDB") 
    .option("dbtable", "transaction_flow") 
    .option("user", "habr") 
    .option("password", "habr12345") 
    .save()

AWS RDS ਨਾਲ ਕੁਨੈਕਸ਼ਨ ਸਥਾਪਤ ਕਰਨ ਬਾਰੇ ਕੁਝ ਸ਼ਬਦ। ਅਸੀਂ "AWS PostgreSQL ਦੀ ਤੈਨਾਤੀ" ਪੜਾਅ 'ਤੇ ਇਸਦੇ ਲਈ ਉਪਭੋਗਤਾ ਅਤੇ ਪਾਸਵਰਡ ਬਣਾਇਆ ਹੈ। ਤੁਹਾਨੂੰ ਡੇਟਾਬੇਸ ਸਰਵਰ url ਦੇ ਤੌਰ 'ਤੇ ਐਂਡਪੁਆਇੰਟ ਦੀ ਵਰਤੋਂ ਕਰਨੀ ਚਾਹੀਦੀ ਹੈ, ਜੋ ਕਨੈਕਟੀਵਿਟੀ ਅਤੇ ਸੁਰੱਖਿਆ ਸੈਕਸ਼ਨ ਵਿੱਚ ਪ੍ਰਦਰਸ਼ਿਤ ਹੁੰਦਾ ਹੈ:

ਸਪਾਰਕ ਅਤੇ ਕਾਫਕਾ ਨੂੰ ਸਹੀ ਢੰਗ ਨਾਲ ਜੋੜਨ ਲਈ, ਤੁਹਾਨੂੰ ਆਰਟੀਫੈਕਟ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਸਮਾਰਕ-ਸਬਮਿਟ ਰਾਹੀਂ ਕੰਮ ਚਲਾਉਣਾ ਚਾਹੀਦਾ ਹੈ ਸਪਾਰਕ-ਸਟ੍ਰੀਮਿੰਗ-ਕਾਫਕਾ-0-8_2.11. ਇਸ ਤੋਂ ਇਲਾਵਾ, ਅਸੀਂ PostgreSQL ਡੇਟਾਬੇਸ ਨਾਲ ਇੰਟਰੈਕਟ ਕਰਨ ਲਈ ਇੱਕ ਆਰਟੀਫੈਕਟ ਦੀ ਵਰਤੋਂ ਵੀ ਕਰਾਂਗੇ; ਅਸੀਂ ਉਹਨਾਂ ਨੂੰ --packages ਰਾਹੀਂ ਟ੍ਰਾਂਸਫਰ ਕਰਾਂਗੇ।

ਸਕ੍ਰਿਪਟ ਦੀ ਲਚਕਤਾ ਲਈ, ਅਸੀਂ ਇਨਪੁਟ ਪੈਰਾਮੀਟਰਾਂ ਦੇ ਰੂਪ ਵਿੱਚ ਸੁਨੇਹਾ ਸਰਵਰ ਦਾ ਨਾਮ ਅਤੇ ਉਹ ਵਿਸ਼ਾ ਵੀ ਸ਼ਾਮਲ ਕਰਾਂਗੇ ਜਿਸ ਤੋਂ ਅਸੀਂ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰਨਾ ਚਾਹੁੰਦੇ ਹਾਂ।

ਇਸ ਲਈ, ਇਹ ਸਿਸਟਮ ਦੀ ਕਾਰਜਕੁਸ਼ਲਤਾ ਨੂੰ ਲਾਂਚ ਕਰਨ ਅਤੇ ਜਾਂਚਣ ਦਾ ਸਮਾਂ ਹੈ:

spark-submit 
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2,
org.postgresql:postgresql:9.4.1207 
spark_job.py localhost:9092 transaction

ਸਭ ਕੁਝ ਕੰਮ ਕੀਤਾ! ਜਿਵੇਂ ਕਿ ਤੁਸੀਂ ਹੇਠਾਂ ਦਿੱਤੀ ਤਸਵੀਰ ਵਿੱਚ ਦੇਖ ਸਕਦੇ ਹੋ, ਜਦੋਂ ਐਪਲੀਕੇਸ਼ਨ ਚੱਲ ਰਹੀ ਹੈ, ਨਵੇਂ ਐਗਰੀਗੇਸ਼ਨ ਨਤੀਜੇ ਹਰ 2 ਸਕਿੰਟਾਂ ਵਿੱਚ ਆਉਟਪੁੱਟ ਹੁੰਦੇ ਹਨ, ਕਿਉਂਕਿ ਜਦੋਂ ਅਸੀਂ ਸਟ੍ਰੀਮਿੰਗ ਕੰਟੈਕਸਟ ਆਬਜੈਕਟ ਬਣਾਉਂਦੇ ਹਾਂ ਤਾਂ ਅਸੀਂ ਬੈਚਿੰਗ ਅੰਤਰਾਲ ਨੂੰ 2 ਸਕਿੰਟਾਂ 'ਤੇ ਸੈੱਟ ਕਰਦੇ ਹਾਂ:

ਅੱਗੇ, ਅਸੀਂ ਸਾਰਣੀ ਵਿੱਚ ਰਿਕਾਰਡਾਂ ਦੀ ਮੌਜੂਦਗੀ ਦੀ ਜਾਂਚ ਕਰਨ ਲਈ ਡੇਟਾਬੇਸ ਲਈ ਇੱਕ ਸਧਾਰਨ ਪੁੱਛਗਿੱਛ ਕਰਦੇ ਹਾਂ ਲੈਣ-ਦੇਣ_ਪ੍ਰਵਾਹ:

ਸਿੱਟਾ

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

ਤੁਸੀਂ ਮੇਰੇ ਰਿਪੋਜ਼ਟਰੀ ਵਿੱਚ ਪੂਰਾ ਸਰੋਤ ਕੋਡ ਇੱਥੇ ਲੱਭ ਸਕਦੇ ਹੋ GitHub.

ਮੈਂ ਇਸ ਲੇਖ 'ਤੇ ਚਰਚਾ ਕਰਕੇ ਖੁਸ਼ ਹਾਂ, ਮੈਂ ਤੁਹਾਡੀਆਂ ਟਿੱਪਣੀਆਂ ਦੀ ਉਮੀਦ ਕਰਦਾ ਹਾਂ, ਅਤੇ ਮੈਂ ਸਾਰੇ ਦੇਖਭਾਲ ਕਰਨ ਵਾਲੇ ਪਾਠਕਾਂ ਤੋਂ ਉਸਾਰੂ ਆਲੋਚਨਾ ਦੀ ਵੀ ਉਮੀਦ ਕਰਦਾ ਹਾਂ।

ਮੈਂ ਤੁਹਾਨੂੰ ਸਫਲਤਾ ਦੀ ਕਾਮਨਾ ਕਰਦਾ ਹਾਂ!

ਜ਼ਬੂ. ਸ਼ੁਰੂ ਵਿੱਚ ਇੱਕ ਸਥਾਨਕ PostgreSQL ਡੇਟਾਬੇਸ ਦੀ ਵਰਤੋਂ ਕਰਨ ਦੀ ਯੋਜਨਾ ਬਣਾਈ ਗਈ ਸੀ, ਪਰ AWS ਲਈ ਮੇਰੇ ਪਿਆਰ ਨੂੰ ਦੇਖਦੇ ਹੋਏ, ਮੈਂ ਡੇਟਾਬੇਸ ਨੂੰ ਕਲਾਉਡ ਵਿੱਚ ਭੇਜਣ ਦਾ ਫੈਸਲਾ ਕੀਤਾ. ਇਸ ਵਿਸ਼ੇ 'ਤੇ ਅਗਲੇ ਲੇਖ ਵਿੱਚ, ਮੈਂ ਦਿਖਾਵਾਂਗਾ ਕਿ AWS ਵਿੱਚ AWS Kinesis ਅਤੇ AWS EMR ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ ਉੱਪਰ ਦੱਸੇ ਗਏ ਪੂਰੇ ਸਿਸਟਮ ਨੂੰ ਕਿਵੇਂ ਲਾਗੂ ਕਰਨਾ ਹੈ। ਖ਼ਬਰਾਂ ਦਾ ਪਾਲਣ ਕਰੋ!

ਸਰੋਤ: www.habr.com