Benötigen Sie Hilfe Konfigurieren von NiSons JsonRecordSetWriter und AvroSchemaRegistry

Question

Ich erstelle ein NiFi WorkFlow, um CSV in JSON zu konvertieren, und ich brauche Hilfe beim Konfigurieren des JsonRecordSetWriter Controller-Dienstes von ConvertRecords. Was passiert ist, dass eine SchemaNotFoundException ausgelöst wird, die besagt: "Schema mit dem Namen 'ccr' kann nicht gefunden werden (Der Name, den ich für das Schema gewählt habe). Das Schema wird aus dem Header im CSV-Dokument mit" InferAvroSchema "und" UpdateAttribute "abgeleitet "ist so konfiguriert, dass ein zugehöriger benannter" schema.name "hinzugefügt wird, der auf 'ccr' gesetzt ist (gemäß Anleitung aus anderen Anleitungen und Anleitungen). Der JsonRecordSetWriter ist so konfiguriert, dass er den Controller-Service" AvroSchemaRegistry "mit einer hinzugefügten Eigenschaft verwendet benannte "ccr" und der Wert für diese Eigenschaft ist auf "$ {inferred.avro.schema}" gesetzt. Ich möchte das abgeleitete Schema, das im Attribut "inferred.avro.schema" enthalten ist, verwendet werden, anstatt es zu müssen Geben Sie den tatsächlichen Text des avro-Schemas als Wert für diese hinzugefügte Eigenschaft an.InferAvroSchemas Eigenschaft SchemaOutputDestination wird auf den Wert "flowfile-attribute" gesetzt, was bedeutet, dass das abgeleitete avro-Schema in ein Attribut namens "inferred.avro.schema" eingefügt wird brauche wirklich Hilfe, die ich nicht kann Spezifizieren Sie das Schema als Text aufgrund der Anforderung des Projekts. Stattdessen möchte ich das abgeleitete Schema verwenden, um CSV-Dateien mit unterschiedlichen Kopf- und Dateninhalten im selben Workflow verarbeiten zu können. Jede Hilfe und Anleitung, die Sie mit mir teilen können, würde ich sehr schätzen.

Answer 1

Wenn Sie einen Leser oder Schreiber mit Schema Access Strategy auf "Schemaname" eingestellt haben, muss das Schema nach Namen aus einer Schemaregistrierung abgerufen werden, und die Schemaregistrierung weiß nichts über $ {inferred.avro .schema}, das sich in einer Flussdatei befindet.

Sie können festlegen, dass der Schreiber die Schemazugriffsstrategie von "Schema Text" verwendet und im Schema-Textfeld $ {inferred.avro.schema} eingibt, damit der Schematext dynamisch aus der eingehenden Flussdatei abgerufen wird. Zu diesem Zeitpunkt verwenden Sie basierend auf Ihren Anforderungen keine Schemaregistrierung.

Eine andere Option, die für Sie arbeiten kann ... Wenn Sie in der Version 1.4.0 sind, können Sie InferAvroSchema eliminieren. Sie würden ConvertRecord mit einem CsvReader verwenden und die Schema-Zugriffsstrategie auf "String-Felder aus Header verwenden" setzen, so dass der Leser ein Schema ableitet und dann in Ihrem JsonRecordSetWriter die Schema-Zugriffsstegie auf "Von Reader vererben" setzt, so dass sie dasselbe verwendet Schema vom Leser bestimmt. Die Vererbungsfunktion existiert in früheren Releases nicht, weshalb diese von 1.4.0 abhängig ist.