Jintegra Kafka Streams ma ' Node.js

L-integrazzjoni Kafka Streams ma' Node.js applikazzjoni hija mod qawwi biex tipproċessa u tanalizza d-dejta direttament minn Apache Kafka f'ambjent Node.js. Kafka Streams jgħinuk tibni l-ipproċessar tad-dejta f'ħin reali u tintegraha bla xkiel fl- Node.js applikazzjoni tiegħek. Hawnhekk hawn gwida speċifika dwar kif tikseb dan:

Pass 1: Installa Kafka Streams u KafkaJS

L-ewwel, għandek bżonn tinstalla Kafka Streams u KafkaJS biex tintegra Kafka fl-applikazzjoni tiegħek Node.js. Tista' tuża npm biex tinstalla dawn il-pakketti:

npm install kafka-streams kafkajs

Pass 2: Oħloq a Kafka Stream

Oħloq fl- applikazzjoni Kafka Stream tiegħek billi tuża l- API. Hawn eżempju bażiku tal-ħolqien ta' dejta biex tipproċessa minn wieħed u toħroġ ir-riżultat għal ieħor: Node.js Kafka Streams Kafka Stream topic topic

const { KafkaStreams } = require('kafka-streams');  
const { Kafka } = require('kafkajs');  
  
const kafka = new Kafka({  
  clientId: 'your-client-id',  
  brokers: ['broker1:port1', 'broker2:port2'],  
});  
  
const kafkaStreams = new KafkaStreams({  
  kafka,  
  logLevel: 2, // Level 2 for debug logs  
});  
  
const streamConfig = {  
  'group.id': 'your-group-id',  
  'metadata.broker.list': 'broker1:port1,broker2:port2',  
  'enable.auto.commit': false,  
  'socket.keepalive.enable': true,  
};  
  
const stream = kafkaStreams.getKStream(streamConfig);  
  
stream  
  .from('input-topic')  
  .filter(record => record.value && record.value.length > 0)  
  .map(record =>({  
    key: record.key,  
    value: record.value.toUpperCase(),  
  }))  
  .to('output-topic');  
  
kafkaStreams.start();  

Pass 3: Ipproċessa Dejta

Fl-eżempju ta 'hawn fuq, ħloqna a Kafka Stream biex tisma' d-dejta mill- input-topic, imbagħad ipproċessajna d-dejta billi kkonvertitha kollha f'ittri kbar u nimbotta r-riżultat lejn il- output-topic.

Pass 4: Mexxi l-Applikazzjoni

Fl-aħħarnett, trid tmexxi Node.js l-applikazzjoni tiegħek biex tibda tipproċessa d-dejta minn Kafka Streams.

Innota li fl-eżempju ta 'hawn fuq, għandek bżonn tissostitwixxi valuri bħal your-client-id, broker1:port1, your-group-id, input-topic u output-topic  bid-dettalji speċifiċi tal-proġett tiegħek.

L-integrazzjoni Kafka Streams ma' Node.js applikazzjoni tippermettilek tibni b'mod flessibbli u qawwi kapaċitajiet ta' pproċessar tad-dejta f'ħin reali.