Source | bact�ries Alice

Zebian | Carol

Photo de couverture | RPSC t�l�charger la Chine visuelle

Produit | RPSC (ID: CSDNnews)

Je crois que beaucoup de petits partenaires ont contact� SparkStreaming, la th�orie ne dit pas trop, aujourd'hui vise principalement � mettre � l'int�gration de Kafka est SparkStreaming tutoriel.

Ce document contient le code, les amis int�ress�s peuvent essayer de mains r�plicats!

Kafka examen

Avant le d�but officiel, passons en revue une vague de Kafka.

Le concept de base illustration

courtier: Services d'installation de la machine Kafka est un courtier

producteur: producteur de message, responsable de l'�criture des donn�es au courtier (push) consommation: Nouvelles des consommateurs, responsable de la prise de donn�es (pull) de kafka traction, l'ancienne version de la n�cessit� des consommateurs � compter ZK, la nouvelle version ne n�cessite pas sujet: Th�me, est l'�quivalent d'une cat�gorie de donn�es, les donn�es stock�es dans diff�rentes entreprises dans diff�rents sujet - le sujet: Differentiated Services La r�plication: Une copie des donn�es stock�es nombre de copies (pour assurer que les donn�es ne sont pas perdues) - Copie: s�curit� des donn�es partition: Partition est une partition physique, une partition est un fichier, un sujet peut avoir 1 ~ n partitions, chaque partition a sa propre copie - Subdivision: lecture et �criture concurrente Groupe consommateurs: Les groupes de consommateurs, un sujet peut avoir plusieurs consommateurs / groupes de consommateurs en m�me temps, plus les consommateurs si un groupe de consommateurs, ils ne peuvent pas �tre donn�es r�p�t�es de consommation - groupes de consommateurs: la vitesse d'augmentation des d�penses de consommation, la gestion unifi�e pratique Remarque: Sujet peut �tre plus qu'un abonnement � la consommation ou d'un groupe, un consommateur / groupe peut �galement vous abonner aux sujets multiples Remarque: Lire les donn�es ne peuvent �tre lues � partir du chef, les donn�es d'�criture ne peuvent �tre �crites au leader, Suiveur viendra faire une copie des donn�es Synchronisez Leader de l�! ! !

Common Commandes

D�marrer kafka

/export/servers/kafka/bin/kafka-server-start.sh -daemon /export/servers/kafka/config/server.properties

arr�t kafka

/export/servers/kafka/bin/kafka-server-stop.sh

Afficher des informations sur le sujet

/export/servers/kafka/bin/kafka-topics.sh --list --zookeeper node01: 2181

rubrique Cr�ation

/export/servers/kafka/bin/kafka-topics.sh --create --zookeeper node01: 2181 --replication-facteur 3 --partitions 3 essais --topic

Afficher des informations sur un sujet

/export/servers/kafka/bin/kafka-topics.sh --describe --zookeeper node01: 2181 Test --topic

Supprimer le sujet

/export/servers/kafka/bin/kafka-topics.sh --zookeeper node01: 2181 Test --delete --topic

producteurs d�part - les producteurs sont g�n�ralement utilis�s pour tester la console

/export/servers/kafka/bin/kafka-console-producer.sh --broker liste node01: 9092 --topic spark_kafka

D�but de consommateurs - les consommateurs sont g�n�ralement utilis�s pour tester la console

/export/servers/kafka/bin/kafka-console-consumer.sh --zookeeper node01: 2181 --topic spark_kafka - de-d�but

Les consommateurs connect�s � l'adresse Borker

/export/servers/kafka/bin/kafka-console-consumer.sh --bootstrap-serveur node01: 9092, node02: 9092, node03: 9092 --topic spark_kafka --from-d�but

L'int�gration KAFKA deux modes expliqu�s

Il est �galement un visage de questions hot spot.

D�veloppement, nous utilisons souvent SparkStreaming donn�es en temps r�el est lu et trait� dans kafka, apr�s la version de spark1.3, kafkaUtils qui fournit deux fa�ons de cr�er DSTREAM de:

1, le mode de r�ception du r�cepteur:

KafkaUtils.createDstream (Sans d�veloppement, nous pouvons comprendre, mais l'entretien peut se demander).
run r�cepteur comme Ex�cuteur r�sident des t�ches en attente dans les donn�es, mais une faible efficacit� du r�cepteur, la n�cessit� d'ouvrir plusieurs, puis fusionner manuellement les donn�es (syndicat), puis trait�es, beaucoup de probl�mes
R�cepteur machine qui a raccroch�, les donn�es peuvent �tre perdues, il est donc n�cessaire d'ouvrir WAL (de WAL) pour assurer la s�curit� des donn�es, l'efficacit� r�duit!
Le r�cepteur est reli� par l'interm�diaire zookeeper file d'attente Kafka, les appels API Kafka d'ordre sup�rieur, le d�calage stock�e dans le zookeeper, maintenu par le r�cepteur.
la consommation d'�tincelle de temps afin d'assurer que les donn�es ne sont pas perdues dans le Checkpoint gardera une copie de l'offset, les donn�es peuvent sembler incoh�rent
Donc, peu importe de quel point de vue, le mode r�cepteur ne sont pas adapt�s pour une utilisation dans le d�veloppement, il a �t� �limin�

2, connexion directe directe

KafkaUtils.createDirectStream (Utilis� dans le d�veloppement, n�cessaires � la ma�trise)
Le mode direct est directement connect� aux partitions kafka d'acquisition de donn�es, chaque donn�e de partition en lecture directement � partir de la capacit� grandement am�lior�e de parall�le
Direct Invoke API de bas niveau Kafka (API sous-jacente), offset leur propre stockage et la maintenance, par d�faut par le poste de contr�le de maintenance Spark, �liminant ainsi incompatible avec ZK
Bien s�r, vous pouvez poss�der la maintenance manuelle, l'existence mysql offset, en Redis
Il peut �tre utilis� dans le d�veloppement bas� sur le mode direct et le mode direct par les caract�ristiques de fonctionnement manuel + des donn�es pr�cises exactement une fois de plus

R�sum�:

mode de r�ception R�cepteur

R�cepteur accepte le risque de multiples rendement �lev� de donn�es, mais il y a des donn�es manquantes

Activer le journal (WAL) pour �viter la perte de donn�es, mais la faible efficacit� de l'�criture deux fois les donn�es.

Zookeeper sauvegarde des donn�es des consommateurs compensent la duplication possible.

Utiliser l'API de haut niveau

Direct Direct Connect

R�cepteur non utilis�, directement � la lecture de kafka partition de donn�es

Ne pas utiliser le m�canisme log (WAL)

Spark maintenir leur propre d�calage

Utilisez API de bas niveau

Extended: A propos de la s�mantique du message

Remarque:

le d�veloppement Kafka SparkStreaming et de l'int�gration en deux versions: 0,8 et 0.10+

Il r�cepteur version 0.8 et le mode direct (version 0.8 mais plus l'environnement de production ne sont pas pris en charge dans la version 0.8 apr�s Spark2.3 a).

0,10 Apr�s avoir retenu que le mode direct (mode Reveiver ne convient pas pour les environnements de production), et la version 0.10 du changement API (plus puissant)

conclusion:

Notre apprentissage et le d�veloppement directement en utilisant la version 0.10 du mod�le direct, mais sur la diff�rence entre le r�cepteur direct et quand l'entrevue devait �tre en mesure de r�pondre en

�tincelle streaming kafka-0-8 (comprendre)

1.Receiver

r�cepteurs KafkaUtils.createDstream utilis�s pour recevoir des donn�es, l'utilisation des niveaux �lev�s de consommation Kafka api, compens�e par le r�cepteur de l'entretien, les donn�es re�ues pour tous les r�cepteurs seront enregistr�s dans Huissiers Spark, suivi par Spark travail en continu pour commencer le traitement de ces donn�es, seront perdues par d�faut, pour activer le journal WAL, il est synchronis� avec les donn�es re�ues stock�es sur un syst�me de fichiers distribu� tels que HDFS. Assurez-vous que les donn�es en cas d'erreur peuvent �tre r�cup�r�s. Bien que cette approche peut �tre combin� avec le m�canisme WAL pour assurer aucune perte et une grande fiabilit� des donn�es, mais l'efficacit� a permis WAL sera plus faible, et ne peut pas garantir que les donn�es sont trait�es une fois et une seule fois, peut �tre trait�e deux fois. Parce qu'il ne peut pas �tre synchronis�s entre le Spark et ZooKeeper.

(Approche maintenant int�gr�e officiellement n'est pas recommand�.)

pr�parations

1) groupe Zookeeper D�marrer

zkServer.sh d�but

2) D�marrer kafka groupe

kafka-server-start.sh /export/servers/kafka/config/server.properties

3. Cr�er un sujet

kafka-topics.sh --create --zookeeper node01: 21811 --partitions --replication facteur 3 --topic spark_kafka

4. Envoyer un message � l'interpr�teur de commandes par th�me

kafka-console-producer.sh --broker liste node01: 9092 --topic spark_kafka

5. Ajouter la kafka d�pendante pom

< d�pendance > < groupId > org.apache.spark < / GroupId > < artifactId > �tincelle streaming kafka-0-8_2.11 < / ArtifactId > < version > 2.2.0 < / version > < / d�pendance >

API

Acquis par le r�cepteur dans le r�cepteur des donn�es de sujet kafka peuvent fonctionner en parall�le r�cepteur plus lit le sujet kafak de donn�es, ici 3

�val receiverDStream: immutable.IndexedSeq = (1 � 3) .map (x = > �{ flux val: ReceiverInputDStream = KafkaUtils.createStream (ssc, zkQuorum, groupId, sujets) ruisseau })

Si la WAL (spark.streaming.receiver.writeAheadLog.enable = true) peut �tre r�gl� pour activer le niveau de stockage (par d�faut StorageLevel.MEMORY_AND_DISK_SER_2)

code d�montre

org.apache.spark.streaming.dstream d'importation. {DSTREAM, ReceiverInputDStream} org.apache.spark.streaming.kafka.KafkaUtils d'importation org.apache.spark.streaming d'importation. {secondes, StreamingContext} org.apache.spark d'importation. {SparkConf, SparkContext} importation scala.collection.immutable objet SparkKafka { def principaux (args: Array): Unit� = { 1 // Cr�ation StreamingContext config val: SparkConf = nouvelle SparkConf.setAppName ( "SparkStream"). setMaster ( "local") .set ( "spark.streaming.receiver.writeAheadLog.enable", "true") // ouvert WAL WAL pour assurer la fiabilit� de la source de donn�es val sc = new SparkContext (config) sc.setLogLevel ( "WARN") val ssc = new StreamingContext (sc, secondes (5)) ssc.checkpoint ( "./ kafka") // ============================================== // 2. Pr�parer les param�tres de configuration val zkQuorum = "node01: 2181, node02: 2181, node03: 2181" val groupId = "�tincelle" val = sujets Plan ( "spark_kafka" - > �2) // 2 repr�sentent chacun le sujet correspond � partition utilise deux fils de passer, RDD kafka des partitions et des partitions // ssc le sujet n'est pas le m�me, ce qui augmente le nombre de threads de consommation, ne pas augmenter le nombre de traitement parall�le des �tincelles de donn�es // 3. Kafka acquis par le r�cepteur de r�cepteur de donn�es sujet peut fonctionner plus r�cepteur parall�le lit le sujet kafak de donn�es, ici 3 val receiverDStream: immutable.IndexedSeq = (1 � 3) .map (x = > �{ flux val: ReceiverInputDStream = KafkaUtils.createStream (ssc, zkQuorum, groupId, sujets) ruisseau }) // 4. m�thode de l'Union utilis�, tout le r�ceptacle r�cepteur DSTREAM sont combin�s pour produire val allDStream: DSTREAM = ssc.union (receiverDStream) // 5. Les donn�es acquises sujet (String, String) Repr�sentation de cha�ne d'un nom de sujet et deuxi�me donn�es repr�sentatives de sujet Cha�ne val donn�es: DSTREAM = allDStream.map (_._ 2) // ============================================== //6.WordCount mots val: DSTREAM = data.flatMap (. _ split ( "")) val wordAndOne: DSTREAM = words.map ((_, 1)) R�sultat val: DSTREAM = wordAndOne.reduceByKey (+ _ _) result.print ssc.start ssc.awaitTermination } }

2.Direct

mode direct v�rifiera p�riodiquement le sujet de la partition correspondante kafka les derni�res donn�es de d�calage trait�es dans chaque lot, puis � l'int�rieur d'une plage de d�calage, plage de lecture API Spark en invoquant simple des donn�es de consommation kafka .

carence directe ne repose pas les outils de surveillance de kafka de Zookeeper
La comparaison directe approche fond�e sur le r�cepteur pr�sente plusieurs avantages:

Simplifier parall�le

Kafka besoin de cr�er plusieurs flux d'entr�e, et leur union, sparkStreaming sera cr�� et le nombre de partitions kafka RDD nombre de partitions diff�rentes, et de lire les donn�es du kafka parall�le, �tincelle RDD le nombre de partitions et les partitions kafka les donn�es sont relation un � un.

efficace

R�cepteur atteindre z�ro perte de donn�es sont les donn�es pr�c�demment stock�es dans WAL, les donn�es sont copi�es � nouveau, elle se traduira par des donn�es copi�es deux fois, le premier �tant copi� kafka, et une fois �crit au WAL. Le Direct ne pas utiliser ce WAL �limine probl�me.

Exactement une fois exactement la s�mantique (-once-s�mantique)

les donn�es du r�cepteur est lu par kafka kafka api de haut niveau pour compenser Zookeeper d'�criture, bien que cette m�thode peut enregistrer les donn�es dans le WAL assurer que les donn�es ne sont pas perdues, mais l'�cart peut �tre enregistr� car le sparkStreaming et ZK d�calage les donn�es de cause � consommer plusieurs fois.

Direct-les exactement une fois s�mantique (EOS) kafka obtenus par api bas niveau, seul le d�calage est stock� dans le poste de contr�le ssc, ce qui �limine les incoh�rences et ssc ZK probl�mes de d�calage.

API

KafkaUtils.createDirectStream (ssc, kafkaParams, sujets)

code d�montre

importation kafka.serializer.StringDecoder org.apache.spark.streaming.dstream d'importation. {DSTREAM, InputDStream} org.apache.spark.streaming.kafka.KafkaUtils d'importation org.apache.spark.streaming d'importation. {secondes, StreamingContext} org.apache.spark d'importation. {SparkConf, SparkContext} SparkKafka2 objet { def principaux (args: Array): Unit� = { 1 // Cr�ation StreamingContext config val: SparkConf = nouvelle SparkConf.setAppName ( "SparkStream"). setMaster ( "local") val sc = new SparkContext (config) sc.setLogLevel ( "WARN") val ssc = new StreamingContext (sc, secondes (5)) ssc.checkpoint ( "./ kafka") // ============================================== // 2. Pr�parer les param�tres de configuration val kafkaParams = Carte ( "metadata.broker.list" - > �"Node01: 9092, node02: 9092, node03: 9092", "group.id" - > �"Spark") sujets val = Set ( "spark_kafka") val allDStream: InputDStream = KafkaUtils.createDirectStream (ssc, kafkaParams, sujets) // 3. Le sujet des donn�es acquises val donn�es: DSTREAM = allDStream.map (_._ 2) // ============================================== // WordCount mots val: DSTREAM = data.flatMap (. _ split ( "")) val wordAndOne: DSTREAM = words.map ((_, 1)) R�sultat val: DSTREAM = wordAndOne.reduceByKey (+ _ _) result.print ssc.start ssc.awaitTermination } }

�tincelle streaming kafka-0-10

explication

�tincelle streaming kafka-0-10 la version, il y a quelques modifications de l'API, un fonctionnement plus souple, utilis� dans le d�veloppement

pom.xml

< ! - < d�pendance > < groupId > org.apache.spark < / GroupId > < artifactId > �tincelle streaming kafka-0-8_2.11 < / ArtifactId > < version > $ {} Spark.version < / version > < / d�pendance > - > < d�pendance > < groupId > org.apache.spark < / GroupId > < artifactId > �tincelle streaming kafka-0-10_2.11 < / ArtifactId > < version > $ {} Spark.version < / version > < / d�pendance >

API:

rubrique Cr�ation

/export/servers/kafka/bin/kafka-topics.sh --create --zookeeper node01: 2181 --replication facteur 3 --partitions 3 --topic spark_kafka

Producteur d�but

/export/servers/kafka/bin/kafka-console-producer.sh --broker liste node01: 9092, node01: 9092, node01: 9092 --topic spark_kafka

code d�montre

importation org.apache.kafka.clients.consumer.ConsumerRecord importation org.apache.kafka.common.serialization.StringDeserializer org.apache.spark.streaming.dstream d'importation. {DSTREAM, InputDStream} org.apache.spark.streaming.kafka010 d'importation. {ConsumerStrategies, KafkaUtils, LocationStrategies} org.apache.spark.streaming d'importation. {secondes, StreamingContext} org.apache.spark d'importation. {SparkConf, SparkContext} objet SparkKafkaDemo { def principaux (args: Array): Unit� = { 1 // Cr�ation StreamingContext //spark.master devrait �tre d�fini comme local, n > �1 val = new SparkConf.setAppName conf ( "wc"). setMaster ( "local") val sc = new SparkContext (conf) sc.setLogLevel ( "WARN") val ssc = new StreamingContext (sc, secondes (5)) // 5 repr�sente 5 secondes pour former un RDD de donn�es de segmentation // pr�t � se connecter les param�tres de Kafka val kafkaParams = Carte ( "Bootstrap.servers" - > �"Node01: 9092, node02: 9092, node03: 9092", "Key.deserializer" - > �classof, "Value.deserializer" - > �classof, "Group.id" - > �"SparkKafkaDemo", // plus t�t: Lorsque d�calage a �t� soumis dans le cadre du district, du d�but des d�penses de d�calage soumis, aucune compensation lorsqu'il est soumis, la consommation de z�ro En aucun donn�es de d�calage lorsqu'il est soumis, un nouveau consommateur cr�� la partition, quand il est d�cal� soumis dans le cadre du district, par rapport aux d�penses de d�marrage soumis offset: // derni�re // none: sujet quand il y a district d�calage a �t� soumis, le d�calage par rapport au d�but du consommateur, tant qu'il est une partition n'existe d�calage soumis pas, une exception est lev�e // cette derni�re configuration automatiquement remis � z�ro d�calage du dernier offset, qui est, s'il y a une position d�cal�e d�cal�e par rapport au d�but de la consommation, il n'y a pas de d�calage aux donn�es de nouvelles d�penses de d�marrage "Auto.offset.reset" - > �� Les derni�res �, // faux moyens ferm�s automatiquement soumis. Par l'�tincelle pour vous aider � soumettre programmeur Checkpoint ou manuel d'entretien "Enable.auto.commit" - > �(Faux: java.lang.Boolean) ) sujets val = Array ( "spark_kafka") // 2. Kafak connexion de donn�es acquises � l'aide KafkaUtil val recordDStream: InputDStream = KafkaUtils.createDirectStream (ssc, LocationStrategies.PreferConsistent, // emplacement strat�gie Source fortement recommand� d'utiliser cette strat�gie, faire spark Ex�cuteur et m�me la correspondance de Kafka Broker ConsumerStrategies.Subscribe (sujets, kafkaParams)) // strat�gie de consommateur, le code source est fortement recommand� d'utiliser cette strat�gie // 3. Valeur d'acquisition de donn�es val lineDStream: DSTREAM = recordDStream.map (. _ valeur) // _ fait r�f�rence ConsumerRecord val wrodDStream: DSTREAM = lineDStream.flatMap // _ fait r�f�rence aux cheveux sur la valeur, � savoir, une ligne de donn�es (_ split ( "").) val wordAndOneDStream: DSTREAM = wrodDStream.map ((_, 1)) R�sultat val: DSTREAM = wordAndOneDStream.reduceByKey (+ _ _) result.print ssc.start // ouvert ssc.awaitTermination // attendre arr�t �l�gant } }

Eh bien, le processus d'int�gration SparkStreaming Kafka dans ce chapitre montrent pour expliquer et vous am�ne � revoir les bases d'une vague de Kafka, si utile pour vous, les mains mal � la main le point d'une � observation �, il ~

Cet article par les auteurs � partir RPSC Blog, le lien d'origine:

https://blog.csdn.net/weixin_44318830/article/details/105612516

chiffre d'affaires flamb�e de 30 ans: d'une communaut� libre � une entreprise de plusieurs milliards de dollars

comprendre l'une des plus grandes r�alisations de l'IA: limitations convolution r�seau de neurones

GitHub a jou� 10000 +, projet de haut niveau Apache ShardingSphere de The Open Road

HKUST Acad�micien interrogation future Zheng Guangting, a r�v�l� les derni�res applications et la pratique de l'intelligence artificielle

intelligents d'exploitation et d'entretien des d�fis en grande promotion: comment Ali r�sista les � doubles 11 chats fin �?

Ethernet Place 2.0 Jeu de garde et mettre en uvre MPC

tr�s difficile pour vous d'�crire neuf questions face de MySQL, nous vous recommandons de la collection!

Route de la soie

Apprenez � conna�tre la Chine

Ultra-d�taill�! SparkStreaming Article d�taill� sur la fa�on d'int�grer Kafka! Peut �tre pratiqu� avec le code

L'int�gration KAFKA deux modes expliqu�s

2.Direct

�tincelle streaming kafka-0-10