Récupérer le Timestamp de la base de données à partir de Kafka
Comment puis-je obtenir les messages ou les données de l'Kafka cluster pour un jour spécifique de la semaine. Par exemple, le 13 septembre, quelqu'un peut-il me fournir le code pour cela. J'ai googlé et trouvé seulement la théorie, mais je veux le code
Il n'existe pas de méthode d'accès pour cela. Aussi, avant de Kafka v0.10 messages ne contiennent pas toutes les informations d'horodatage, il est donc impossible de savoir quand un message a été écrit dans une rubrique.
De Kafka v0.10 chaque message contient un méta-données d'horodatage de l'attribut, qui est défini par le producteur sur la création de messages de temps, ou par le courtier sur le message de l'insertion. Un index est prévu, mais pas encore disponible. Ainsi, vous avez besoin de consommer la totalité du sujet et de vérifier le champ d'horodatage (et ignorer tous les messaged vous n'êtes pas intéressés). Pour trouver le début, vous pouvez aussi faire une recherche binaire à l'égard des décalages et les horodatages pour trouver le premier message plus rapidement.
Mise à jour:
Kakfa 0.10.1 ajoute un index. Il permet de seek pour le premier enregistrement avec un timestamp est égale à ou plus grande de l'horodatage. Vous pouvez l'utiliser via KafkaConsumer#offsetsForTime(). Ce sera le retour de la correspondante de décalages et vous pouvez les nourrir en KafkaConsumer#seek(). Vous pouvez seulement de consommer les données et vérifier les dossiers champ d'horodatage via ConsumerRecord#timestamp() à voir quand vous pouvez arrêter le traitement.
Noter que les données sont strictement commandé par décalages, mais pas par l'horodateur. Ainsi, durant le traitement, vous pourriez obtenir la "fin" des enregistrements avec petits horodatage de votre horodatage de départ (vous pourriez simple d'ignorer ces dossiers bien que).
Un problème plus difficile à résoudre arrive en retard enregistrer à la fin de votre intervalle de recherche. Après avoir obtenu le premier horodatage avec une plus grande timestamp que votre intervalle de recherche, il y a encore des enregistrements avec horodatage qui font partie de votre intervalle de recherche plus tard (si ces documents ne suis ajouté à la rubrique "en retard"). Il n'y a aucun moyen de savoir que, bien que. Donc, vous voudrez peut-être garder à la lecture de "un peu plus" dossiers et vérifier si il y a la "fin" des dossiers. Comment beaucoup de "certains dossiers", est une décision que vous devez faire par vous-même.
Il n'y a pas de ligne directrice générale -- si vous avez des connaissances supplémentaires sur votre écriture "pattern", il peut aider à définir une bonne stratégie pour le nombre d'enregistrements que vous souhaitez consommateur après votre intervalle de recherche "terminé". Bien sûr, il y a deux stratégies par défaut: (1) arrêt à la très premier enregistrement avec les plus grandes timestamp que vous l'intervalle de recherche (et effectivement ignorer tout retard de l'arrivée des documents, si vous utilisez le "journal d'ajouter de temps de configuration" c'est bien sûr, un coffre-fort à la stratégie); (2) vous avez lu jusqu'à la fin du journal -- c'est le plus sûr stratégie à l'égard de l'exhaustivité, mais peut entraîner prohibitif des frais généraux (également noter que l'enregistrement peut être ajouté à tout moment et si l'enregistrement de "retard" peut être arbitrairement grand, une fin d'enregistrement peut-être même ajouter après vous arrivez en fin de journal).
Dans la pratique, il pourrait être une bonne idée de réfléchir à un "max de retard prévu" et de lire jusqu'à ce que vous obtenir un record avec plus d'horodatage de cette limite supérieure de retard lié.
"de Kafka v0.10 chaque message contient un méta-données d'horodatage de l'attribut, qui est défini par le producteur sur la création de messages de temps, ou par le courtier sur le message de l'insertion." Pouvez-vous expliquer la façon dont le timestamp est défini par le producteur? Est-il un moyen d'avoir toujours le serveur pour fournir le timestamp? ProducerRecord a plusieurs surcharge pour son constructeur; certains acceptent un paramètre timestamp (type long). Pour le courtier côté d'horodateurs, vous avez besoin de changer le sujet correspondant config message.timestamp.type cf kafka.apache.org/documentation/#topicconfigs afin de kafka flux avec horodatage extracteur appliquée, l'horodatage de chaque enregistrement est défini par le timestamp de l'extracteur, c'est bien cela? @MatthiasJ.Sax Dans Kafka cours d'eau, le TimestampExtractor est utilisé pour obtenir les horodateurs pour les enregistrements d'entrée. Sur l'écriture, l'enregistrement d'horodatage sera définie de manière explicite par les flots. Atm, ce n'est pas le document comment Kafka Flux calcule et détermine le timestamp pour les enregistrements de sortie (nous avons l'intention de combler cette lacune bientôt). Remarque, depuis la version 2.0, vous pouvez définir les horodatages d'un enregistrement dans Kafka Flux explicitement (cwiki.apache.org/confluence/display/KAFKA/...) et ce timestamp serait de définir si le document est écrit à la rubrique de résultat.
l'obtention de données pour un jour spécifique de kafka n'est PAS très efficace, puisque les données sont stockées de façon linéaire à l'intérieur de kafka sur chaque courtier du système de stockage. par conséquent, même si vous avez d'horodatage à l'intérieur de votre message ou de l'utilisation de kafka message métadonnées qui pourraient contenir des timestamp dans la suite de kafka version message(>=0.10), il vous reste à analyser l'ensemble du sujet sur chaque partition pour obtenir les données. en raison du fait que les données à l'intérieur de kafka n'est pas indexé par jour, mais seulement compenser.
rappelez-vous, kafka est une file d'attente, PAS une base de données. si vous souhaitez que cette fonction de la date de récupérer modèle, vous pourriez envisager de stocker kafka message à l'intérieur d'un autre adapté les bases de données système et d'utiliser le timestamp que votre index.
Il n'existe pas de méthode d'accès pour cela. Aussi, avant de Kafka
v0.10messages ne contiennent pas toutes les informations d'horodatage, il est donc impossible de savoir quand un message a été écrit dans une rubrique.De Kafka
v0.10chaque message contient un méta-données d'horodatage de l'attribut, qui est défini par le producteur sur la création de messages de temps, ou par le courtier sur le message de l'insertion. Un index est prévu, mais pas encore disponible. Ainsi, vous avez besoin de consommer la totalité du sujet et de vérifier le champ d'horodatage (et ignorer tous les messaged vous n'êtes pas intéressés). Pour trouver le début, vous pouvez aussi faire une recherche binaire à l'égard des décalages et les horodatages pour trouver le premier message plus rapidement.Mise à jour:
Kakfa
0.10.1ajoute un index. Il permet deseekpour le premier enregistrement avec un timestamp est égale à ou plus grande de l'horodatage. Vous pouvez l'utiliser viaKafkaConsumer#offsetsForTime(). Ce sera le retour de la correspondante de décalages et vous pouvez les nourrir enKafkaConsumer#seek(). Vous pouvez seulement de consommer les données et vérifier les dossiers champ d'horodatage viaConsumerRecord#timestamp()à voir quand vous pouvez arrêter le traitement.Noter que les données sont strictement commandé par décalages, mais pas par l'horodateur. Ainsi, durant le traitement, vous pourriez obtenir la "fin" des enregistrements avec petits horodatage de votre horodatage de départ (vous pourriez simple d'ignorer ces dossiers bien que).
Un problème plus difficile à résoudre arrive en retard enregistrer à la fin de votre intervalle de recherche. Après avoir obtenu le premier horodatage avec une plus grande timestamp que votre intervalle de recherche, il y a encore des enregistrements avec horodatage qui font partie de votre intervalle de recherche plus tard (si ces documents ne suis ajouté à la rubrique "en retard"). Il n'y a aucun moyen de savoir que, bien que. Donc, vous voudrez peut-être garder à la lecture de "un peu plus" dossiers et vérifier si il y a la "fin" des dossiers. Comment beaucoup de "certains dossiers", est une décision que vous devez faire par vous-même.
Il n'y a pas de ligne directrice générale -- si vous avez des connaissances supplémentaires sur votre écriture "pattern", il peut aider à définir une bonne stratégie pour le nombre d'enregistrements que vous souhaitez consommateur après votre intervalle de recherche "terminé". Bien sûr, il y a deux stratégies par défaut: (1) arrêt à la très premier enregistrement avec les plus grandes timestamp que vous l'intervalle de recherche (et effectivement ignorer tout retard de l'arrivée des documents, si vous utilisez le "journal d'ajouter de temps de configuration" c'est bien sûr, un coffre-fort à la stratégie); (2) vous avez lu jusqu'à la fin du journal -- c'est le plus sûr stratégie à l'égard de l'exhaustivité, mais peut entraîner prohibitif des frais généraux (également noter que l'enregistrement peut être ajouté à tout moment et si l'enregistrement de "retard" peut être arbitrairement grand, une fin d'enregistrement peut-être même ajouter après vous arrivez en fin de journal).
Dans la pratique, il pourrait être une bonne idée de réfléchir à un "max de retard prévu" et de lire jusqu'à ce que vous obtenir un record avec plus d'horodatage de cette limite supérieure de retard lié.
ProducerRecorda plusieurs surcharge pour son constructeur; certains acceptent un paramètre timestamp (type long). Pour le courtier côté d'horodateurs, vous avez besoin de changer le sujet correspondant configmessage.timestamp.typecf kafka.apache.org/documentation/#topicconfigsafin de kafka flux avec horodatage extracteur appliquée, l'horodatage de chaque enregistrement est défini par le timestamp de l'extracteur, c'est bien cela? @MatthiasJ.Sax
Dans Kafka cours d'eau, le
TimestampExtractorest utilisé pour obtenir les horodateurs pour les enregistrements d'entrée. Sur l'écriture, l'enregistrement d'horodatage sera définie de manière explicite par les flots. Atm, ce n'est pas le document comment Kafka Flux calcule et détermine le timestamp pour les enregistrements de sortie (nous avons l'intention de combler cette lacune bientôt). Remarque, depuis la version 2.0, vous pouvez définir les horodatages d'un enregistrement dans Kafka Flux explicitement (cwiki.apache.org/confluence/display/KAFKA/...) et ce timestamp serait de définir si le document est écrit à la rubrique de résultat.OriginalL'auteur Matthias J. Sax
l'obtention de données pour un jour spécifique de kafka n'est PAS très efficace, puisque les données sont stockées de façon linéaire à l'intérieur de kafka sur chaque courtier du système de stockage. par conséquent, même si vous avez d'horodatage à l'intérieur de votre message ou de l'utilisation de kafka message métadonnées qui pourraient contenir des timestamp dans la suite de kafka version message(>=0.10), il vous reste à analyser l'ensemble du sujet sur chaque partition pour obtenir les données. en raison du fait que les données à l'intérieur de kafka n'est pas indexé par jour, mais seulement compenser.
rappelez-vous, kafka est une file d'attente, PAS une base de données. si vous souhaitez que cette fonction de la date de récupérer modèle, vous pourriez envisager de stocker kafka message à l'intérieur d'un autre adapté les bases de données système et d'utiliser le timestamp que votre index.
OriginalL'auteur linehrr