मैं NiFi JoltTransformJSON को कई रिकॉर्ड भेज रहा हूं जैसे:

{"uid":"x1","timestamp_ms":1419717816375}
{"uid":"x2","timestamp_ms":1419717816376}
...

वे ComsumeKafkaRecord के माध्यम से आ रहे हैं, और इसका RecordWriter JsonRecordSetWriter "एक लाइन प्रति ऑब्जेक्ट" के साथ कॉन्फ़िगर किया गया है। मैंने डेटा प्रोवेंस के माध्यम से सत्यापित किया, कि जब मैं काफ्का को 5 भेजता हूं, तो उसमें से 5 रिकॉर्ड निकलते हैं।

शिफ्ट ट्रांसफॉर्म के लिए झटका युक्ति है

{
  "*": "&0",
 "timestamp_ms": "timestamp"
}

तो मैं सिर्फ एक क्षेत्र का नाम बदलना चाहता हूँ। मैंने उन्नत अनुभाग में सत्यापित किया है कि यह युक्ति अनुरोध के अनुसार मेरे जेसन को बदल देती है।

लेकिन जब मैं सभी घटकों को शुरू करता हूं और 5 संदेश भेजता हूं, तो मुझे केवल जोल्टट्रांसफॉर्मजेसन से पहला मिलता है - डेटा प्रोवेंस के साथ सत्यापित।

{"uid":"x1","timestamp":1419717816375}

यहाँ क्या गलत है?

अपडेट: यह पता चला कि मुझे JoltTransformRecord का उपयोग करना था, फिर यह मेरी अपेक्षा के अनुरूप चला, यानी 5 संदेश अंदर, 5 आउट।

लेकिन मैं अभी भी हैरान हूं कि JoltTransformJson ने ऐसा क्या व्यवहार किया?

0
Artem Trunov 17 अगस्त 2019, 00:33

1 उत्तर

सबसे बढ़िया उत्तर

यह इस तरह का व्यवहार करने वाला तथ्य यह है कि आपने अपने डेटा (प्रति पंक्ति एक रिकॉर्ड) को JSON के रूप में संदर्भित किया है, जबकि आपका डेटा JSON नहीं है। यदि यह JSON था, तो यह कम से कम [] के साथ शुरू और समाप्त होगा और प्रत्येक पंक्ति (अंतिम एक को छोड़कर) अल्पविराम के साथ समाप्त होगी।

चूंकि JoltTransformJSON केवल JSON स्वीकार करता है, इसलिए उसने ऐसा व्यवहार किया।

आपके संपादन के संबंध में, JoltTransformRecord जानता है कि अलग-अलग JSON ऑब्जेक्ट्स के रूप में रिकॉर्ड्स को व्यक्तिगत रूप से कैसे संसाधित किया जाए, और इसीलिए इसने इसका उपयोग करके काम किया।

2
Ben Yaakobi 18 अगस्त 2019, 11:24