Package org.apache.nifi.processors.standard

Class QueryRecord

java.lang.Object
org.apache.nifi.components.AbstractConfigurableComponent
org.apache.nifi.processor.AbstractSessionFactoryProcessor
org.apache.nifi.processor.AbstractProcessor
org.apache.nifi.processors.standard.QueryRecord
All Implemented Interfaces:
org.apache.nifi.components.ConfigurableComponent, org.apache.nifi.processor.Processor
@SideEffectFree @SupportsBatching @Tags({"sql","query","calcite","route","record","transform","select","update","modify","etl","filter","record","csv","json","logs","text","avro","aggregate"}) @InputRequirement(INPUT_REQUIRED) @CapabilityDescription("Evaluates one or more SQL queries against the contents of a FlowFile. The result of the SQL query then becomes the content of the output FlowFile. This can be used, for example, for field-specific filtering, transformation, and row-level filtering. Columns can be renamed, simple calculations and aggregations performed, etc. The Processor is configured with a Record Reader Controller Service and a Record Writer service so as to allow flexibility in incoming and outgoing data formats. The Processor must be configured with at least one user-defined property. The name of the Property is the Relationship to route data to, and the value of the Property is a SQL SELECT statement that is used to specify how input data should be transformed/filtered. The SQL statement must be valid ANSI SQL and is powered by Apache Calcite. If the transformation fails, the original FlowFile is routed to the \'failure\' relationship. Otherwise, the data selected will be routed to the associated relationship. If the Record Writer chooses to inherit the schema from the Record, it is important to note that the schema that is inherited will be from the ResultSet, rather than the input Record. This allows a single instance of the QueryRecord processor to have multiple queries, each of which returns a different set of columns and aggregations. As a result, though, the schema that is derived will have no schema name, so it is important that the configured Record Writer not attempt to write the Schema Name as an attribute if inheriting the Schema from the Record. See the Processor Usage documentation for more information.") @DynamicRelationship(name="<Property Name>", description="Each user-defined property defines a new Relationship for this Processor.") @DynamicProperty(name="The name of the relationship to route data to", value="A SQL SELECT statement that is used to determine what data should be routed to this relationship.", expressionLanguageScope=FLOWFILE_ATTRIBUTES, description="Each user-defined property specifies a SQL SELECT statement to run over the data, with the data that is selected being routed to the relationship whose name is the property name") @WritesAttribute(attribute="mime.type",description="Sets the mime.type attribute to the MIME Type specified by the Record Writer") @WritesAttribute(attribute="record.count",description="The number of records selected by the query") @WritesAttribute(attribute="QueryRecord.Route",description="The relation to which the FlowFile was routed") @UseCase(description="Filter out records based on the values of the records\' fields",keywords={"filter out","remove","drop","strip out","record field","sql"},configuration="\"Record Reader\" should be set to a Record Reader that is appropriate for your data.\n\"Record Writer\" should be set to a Record Writer that writes out data in the desired format.\n\nOne additional property should be added.\nThe name of the property should be a short description of the data to keep.\nIts value is a SQL statement that selects all columns from a table named `FLOW_FILE` for relevant rows.\nThe WHERE clause selects the data to keep. I.e., it is the exact opposite of what we want to remove.\nIt is recommended to always quote column names using double-quotes in order to avoid conflicts with SQL keywords.\nFor example, to remove records where either the name is George OR the age is less than 18, we would add a property named \"adults not george\" with a value that selects records where the name is not George AND the age is greater than or equal to 18. So the value would be `SELECT * FROM FLOWFILE WHERE \"name\" <> \'George\' AND \"age\" >= 18`\n\nAdding this property now gives us a new Relationship whose name is the same as the property name. So, the \"adults not george\" Relationship should be connected to the next Processor in our flow.\n") @UseCase(description="Keep only specific records",keywords={"keep","filter","retain","select","include","record","sql"},configuration="\"Record Reader\" should be set to a Record Reader that is appropriate for your data.\n\"Record Writer\" should be set to a Record Writer that writes out data in the desired format.\n\nOne additional property should be added.\nThe name of the property should be a short description of the data to keep.\nIts value is a SQL statement that selects all columns from a table named `FLOW_FILE` for relevant rows.\nThe WHERE clause selects the data to keep.\nIt is recommended to always quote column names using double-quotes in order to avoid conflicts with SQL keywords.\nFor example, to keep only records where the person is an adult (aged 18 or older), add a property named \"adults\" with a value that is a SQL statement that selects records where the age is at least 18. So the value would be `SELECT * FROM FLOWFILE WHERE \"age\" >= 18`\n\nAdding this property now gives us a new Relationship whose name is the same as the property name. So, the \"adults\" Relationship should be connected to the next Processor in our flow.\n") @UseCase(description="Keep only specific fields in a a Record, where the names of the fields to keep are known",keywords={"keep","filter","retain","select","include","record","fields","sql"},configuration="\"Record Reader\" should be set to a Record Reader that is appropriate for your data.\n\"Record Writer\" should be set to a Record Writer that writes out data in the desired format.\n\nOne additional property should be added.\nThe name of the property should be a short description of the data to keep, such as `relevant fields`.\nIts value is a SQL statement that selects the desired columns from a table named `FLOW_FILE` for relevant rows.\nThere is no WHERE clause.\nIt is recommended to always quote column names using double-quotes in order to avoid conflicts with SQL keywords.\nFor example, to keep only the `name`, `age`, and `address` fields, add a property named `relevant fields` with a value of `SELECT \"name\", \"age\", \"address\" FROM FLOWFILE`\n\nAdding this property now gives us a new Relationship whose name is the same as the property name. So, the `relevant fields` Relationship should be connected to the next Processor in our flow.\n") @UseCase(description="Route record-oriented data for processing based on its contents",keywords={"record","route","conditional processing","field"},configuration="\"Record Reader\" should be set to a Record Reader that is appropriate for your data.\n\"Record Writer\" should be set to a Record Writer that writes out data in the desired format.\n\nFor each route that you want to create, add a new property.\nThe name of the property should be a short description of the data that should be selected for the route.\nIts value is a SQL statement that selects all columns from a table named `FLOW_FILE`. The WHERE clause selects the data that should be included in the route.\nIt is recommended to always quote column names using double-quotes in order to avoid conflicts with SQL keywords.\n\nA new outbound relationship is created for each property that is added. The name of the relationship is the same as the property name.\n\nFor example, to route data based on whether or not it is a large transaction, we would add two properties:\n`small transaction` would have a value such as `SELECT * FROM FLOWFILE WHERE transactionTotal < 100`\n`large transaction` would have a value of `SELECT * FROM FLOWFILE WHERE transactionTotal >= 100`\n") public class QueryRecord extends org.apache.nifi.processor.AbstractProcessor

  • Field Details

    • ROUTE_ATTRIBUTE_KEY

      public static final String ROUTE_ATTRIBUTE_KEY
      See Also:

    • RECORD_READER_FACTORY

      static final org.apache.nifi.components.PropertyDescriptor RECORD_READER_FACTORY

    • RECORD_WRITER_FACTORY

      static final org.apache.nifi.components.PropertyDescriptor RECORD_WRITER_FACTORY

    • INCLUDE_ZERO_RECORD_FLOWFILES

      static final org.apache.nifi.components.PropertyDescriptor INCLUDE_ZERO_RECORD_FLOWFILES

    • PROPERTY_DESCRIPTORS

      private static final List<org.apache.nifi.components.PropertyDescriptor> PROPERTY_DESCRIPTORS

    • REL_ORIGINAL

      public static final org.apache.nifi.processor.Relationship REL_ORIGINAL

    • REL_FAILURE

      public static final org.apache.nifi.processor.Relationship REL_FAILURE

    • relationships

      private final Set<org.apache.nifi.processor.Relationship> relationships

    • statementQueues

      private final com.github.benmanes.caffeine.cache.Cache<Tuple<String,RecordSchema>,BlockingQueue<QueryRecord.CachedStatement>> statementQueues

  • Constructor Details

    • QueryRecord

      public QueryRecord()

  • Method Details

    • migrateProperties

      public void migrateProperties(PropertyConfiguration config)

    • init

      protected void init(org.apache.nifi.processor.ProcessorInitializationContext context)
      Overrides:
      init in class org.apache.nifi.processor.AbstractSessionFactoryProcessor

    • getRelationships

      public Set<org.apache.nifi.processor.Relationship> getRelationships()
      Specified by:
      getRelationships in interface org.apache.nifi.processor.Processor
      Overrides:
      getRelationships in class org.apache.nifi.processor.AbstractSessionFactoryProcessor

    • getSupportedPropertyDescriptors

      protected List<org.apache.nifi.components.PropertyDescriptor> getSupportedPropertyDescriptors()
      Overrides:
      getSupportedPropertyDescriptors in class org.apache.nifi.components.AbstractConfigurableComponent

    • onPropertyModified

      public void onPropertyModified(org.apache.nifi.components.PropertyDescriptor descriptor, String oldValue, String newValue)
      Specified by:
      onPropertyModified in interface org.apache.nifi.components.ConfigurableComponent
      Overrides:
      onPropertyModified in class org.apache.nifi.components.AbstractConfigurableComponent

    • getSupportedDynamicPropertyDescriptor

      protected org.apache.nifi.components.PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName)
      Overrides:
      getSupportedDynamicPropertyDescriptor in class org.apache.nifi.components.AbstractConfigurableComponent

    • cleanup

      @OnStopped public void cleanup()

    • onCacheEviction

      private void onCacheEviction(Tuple<String,RecordSchema> key, BlockingQueue<QueryRecord.CachedStatement> queue, com.github.benmanes.caffeine.cache.RemovalCause cause)

    • clearQueue

      private void clearQueue(BlockingQueue<QueryRecord.CachedStatement> statementQueue)

    • onTrigger

      public void onTrigger(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session)
      Specified by:
      onTrigger in class org.apache.nifi.processor.AbstractProcessor

    • getStatement

      private QueryRecord.CachedStatement getStatement(String sql, RecordSchema schema, Supplier<QueryRecord.CachedStatement> statementBuilder)

    • buildCachedStatement

      private QueryRecord.CachedStatement buildCachedStatement(String sql, RecordSchema recordSchema)

    • query

      private QueryRecord.QueryResult query(org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, RecordSchema schema, String sql, RecordReaderFactory recordReaderFactory) throws SQLException
      Throws:
      SQLException

    • closeQuietly

      private static void closeQuietly(AutoCloseable... closeables)

    • closeQuietly

      private static void closeQuietly(org.apache.nifi.logging.ComponentLog logger, AutoCloseable... closeables)