apache-nifi - 从流文件内容中提取多行内容

Question

我正在从 MySQL 表导入数据(仅适用于选定的列)并将其放入 HDFS。完成此操作后，我想在 Hive 中创建一个表。

为此，我有一个 schema.sql 文件，其中包含整个表的 CREATE TABLE 语句，我想仅为我导入的列生成新的 CREATE TABLE 语句。

类似于我在下面的示例中使用 grep 所做的事情。

我使用了 FetchFile 和 ExtractText 但无法使其工作。如果我将整体架构放入属性中，如何使用 NiFi 处理器甚至表达式语言来实现此目的？

或者有更好的方法在导入的数据上创建表格吗？

Answer 1

NiFi可以根据流文件内容生成创建表语句

1.Creating ORC tables by using ConvertAvroToORC processor:

• 如果您要将 avro 数据转换为 ORC 格式然后存储到 HDFS 中，则 ConvertAvroToORC 处理器会添加 hive.ddl流文件的属性。

• PutHDFS 处理器添加 absolute.hdfs.path流文件的属性。

• 我们可以使用这个hive.ddl、absolute.hdfs.path属性并在HDFS目录之上动态创建orc表。

流程:

 Pull data from source(ExecuteSQL...etc)
  -> ConvertAvroToORC //add Hive DbName,TableName in HiveTableName property value--> 
  -> PutHDFS //store the orc file into HDFS location --> 
  -> ReplaceText //Replace the flowfile content with ${hive.ddl} Location '${absolute.hdfs.path}'--> 
  -> PutHiveQL //execute the create table statement

引用this有关更多详细信息的链接重新升级到上述流程。

2.Creating Avro tables by using ExtractAvroMetaData processor:

• 在 NiFi 中，一旦我们使用 QueryDatabaseTable、ExecuteSQL 处理器提取数据，数据的格式就是 AVRO。

• 我们可以基于avro架构(.avsc文件)创建Avro表，并使用ExtractAvroMetaData处理器我们可以提取schema 并保留为 flowfile 属性，然后通过使用此 schema 我们可以动态创建 AvroTables。

流程:

ExecuteSQL (success)|-> PutHDFS //store data into HDFS
           (success)|-> ExtractAvroMetadata //configure Metadata Keys as avro.schema 
                     -> ReplaceText //replace flowfile content with avro.schema
                     -> PutHDFS //store the avsc file into schema directory
                     -> ReplaceText //create avro table on top of schema directory
                     -> PutHiveQL //execute the hive.ddl

示例 AVRO创建表语句:

CREATE TABLE as_avro
  ROW FORMAT SERDE
  'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
  STORED as INPUTFORMAT
  'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
  OUTPUTFORMAT
  'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
  TBLPROPERTIES (
    'avro.schema.url'='/path/to/the/schema/test_serializer.avsc');

我们将在上述流程中使用 ReplaceText 处理器来更改架构 URL 的路径。

使用ExecuteSQL处理器获取所有创建表语句(或)来自(sys.tables/INFORMATION_SCHEMA.COLUMNS ..etc)的列信息的另一种方法 从源(如果源系统允许)并将脚本写入 map the data types进入hive appropriate types然后将它们存储在您的desired format中在 hive 中。

编辑:

运行grep我们需要使用ExecuteStreamCommand处理器

对流文件内容执行命令

ESC 配置:

然后喂output stream与 ExtractText 处理器的关系

ET 配置:

添加新属性

内容

(?s)(.*)

然后content attribute添加到流程文件中，您可以使用该属性并准备创建表语句。