NiFi中不同数据源的接入与集成

# 1. NiFi简介和基本概念

## 1.1 NiFi概述
Apache NiFi是一个易于使用、强大而可靠的数据处理和分发系统，用于从各种数据源收集、聚合和传输数据。NiFi提供了直观的用户界面和强大的数据管道概念，使得用户可以轻松地配置、管理和监控数据流。它是一个开源项目，最初由美国国家安全局(NSA)开发，后来捐赠给了Apache软件基金会并成为顶级项目。

## 1.2 数据流和数据管道
NiFi基于数据流概念，数据流由称为处理器的数据处理单元组成，处理器通过连接的关系构成数据管道。数据管道定义了数据流在系统中的传输路线，可以包括数据源、转换、路由、存储等操作。NiFi的数据管道可以实现高度可视化和可配置，用户可以通过图形化界面直观地管理和修改数据流。

## 1.3 NiFi的核心功能和优势
NiFi的核心功能包括数据收集、简单转换、数据路由、数据优先级设置、数据追踪和验证等。其优势在于可视化数据流管理、实时数据传输、数据流的可靠性保证，以及对多种数据类型和格式的全面支持。NiFi还提供了丰富的监控和日志功能，能够帮助用户全面了解和控制数据流的运行状况。

接下来将深入介绍不同数据源的接入方式。

# 2. 不同数据源的接入方式

### 2.1 数据库数据源接入

在NiFi中，我们可以通过配置DBeve Processor来轻松实现对不同数据库的数据抽取和加载。以下是一个Java示例代码：

import org.apache.nifi.components.*;
import org.apache.nifi.dbcp.DBCPService;
import org.apache.nifi.dbcp.hive.*;
import org.apache.nifi.processor.*;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.logging.*;
import org.apache.nifi.annotation.lifecycle.*;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;

@Tags({"DB", "SQL", "Get", "Hive", "Select"})
@CapabilityDescription("Execute provided SQL select query to read data from a Hive database.")


public class GetHiveQL extends AbstractProcessor {

private static List<PropertyDescriptor> properties;
private static Set<Relationship> relationships;
private ComponentLog log;

// Define relationships

public static final Relationship SUCCESS = new Relationship.Builder()
.name("success")
.description("Succesful completion of the processor")
.build();

public static final Relationship FAILURE = new Relationship.Builder()
.name("failure")
.description("Failed to complete the processor")
.build();

@Override

protected void init(final ProcessorInitializationContext context){
List<PropertyDescriptor> properties = new ArrayList<>();
this.properties= Collections.unmodifiableList(properties);
Set<Relationship> relationships = new HashSet<>();
properties.add(SQL_QUERY);
properties.add(dbcpService);
this.relationships= Collections.unmodifiableSet(relationships);
private final static PropertyDescriptor SQL_QUERY = new PropertyDescriptor.Builder()
.name("sql-statement")
.description("SQL select query")
.required(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();

// Load Service
private final static PropertyDescriptor DBCP_SERVICE = new PropertyDescriptor.Builder()
.name("dbcp-service")
.description("Database connection pool service")
.required(true)
.identifiesControllerService(DBCPService.class)
.build();
}


@Override

public void onTrigger(final ProcessContext context, final ProcessSession session){

}

@Override

public Set<Relationship> getRelationships(){
return relationships;
}

@Override

public List<PropertyDescriptor> getSupportedPropertyDescriptors(){
return properties;
}
}

``