Apache Nifi Install作为Ambari服务 - 在配置中找不到配置参数“Kafka_Broker_Hosts”

技术标签： Apache-Kafka. Apache-nifi. HortonWorks-Data-Platform ambari.

我的系统环境如下：

使用带有Kafka / ZK的2个节点HDP 2.5群集。

Node1：Ambari-Server，Ambari-Agent
Node2：Ambari-Agent

我已经设置为Kafka经纪人，其中ZooKeeper服务器在第一个方面运行。我还在节点上成功安装并启动了NiFi。

但是在尝试将Nifi添加为Ambari Service时，Nifi Server安装失败了以下错误（请参阅错误日志:)

我能弄清楚的是失败背后的主要原因是

**resource_management.core.exceptions.Fail: Configuration parameter 'kafka_broker_hosts' was not found in configurations dictionary!**

我已经在/usr/hdp/2.5.0.0-1245/nifi/conf/nifi.properties下检查了，并在建议下，从默认8080到9090更改了Nifi的Web端口

https://community.hortonworks.com/questions/44042/addding-nifi-server-to-hdp-25-sandbox.html.

然后我重新启动了Ambari服务器并停止并启动了NiFi服务，但仍然存在相同的问题。

# web properties #
nifi.web.war.directory=./lib
nifi.web.http.host=
nifi.web.http.port=9090
nifi.web.https.host=
nifi.web.https.port=
nifi.web.jetty.working.directory=./work/jetty
nifi.web.jetty.threads=200

我选中在下面的文件中，检查配置参数并使用“Kafka_Broker_Host”术语搜索

/var/lib/ambari-server/resources/common-services/atlas/0.1.0.3/package/scripts/params.py.

这些是我能弄清楚的相关代号

**245 # ToDo: Kafka port to Atlas
246 # Used while upgrading the stack in a kerberized cluster and running kafka-acls.sh
247 hosts_with_kafka = default('/clusterHostInfo/kafka_broker_hosts', [])
248 host_with_kafka = hostname in hosts_with_kafka**

但是链接 https://issues.apache.org/jira/secure/Attachment/12834513/ambari-16693.patch. 一些不同的代码补丁谈判。

我很困惑，正在努力了解Params.py中需要一些代码更改。 Ambari-16693.patch. 这是Apache Jira，或者是它，以便Apache Nifi完全不兼容为Ambari服务。

错误日志：

stderr: 
Traceback (most recent call last):
  File "/var/lib/ambari-agent/cache/stacks/HDP/2.5/services/NIFI/package/scripts/master.py", line 130, in <module>
    Master().execute()
  File "/usr/lib/python2.6/site-packages/resource_management/libraries/script/script.py", line 280, in execute
    method(env)
  File "/var/lib/ambari-agent/cache/stacks/HDP/2.5/services/NIFI/package/scripts/master.py", line 14, in install
    import params
  File "/var/lib/ambari-agent/cache/stacks/HDP/2.5/services/NIFI/package/scripts/params.py", line 90, in <module>
    kafka_broker_host = str(master_configs['kafka_broker_hosts'][0])
  File "/usr/lib/python2.6/site-packages/resource_management/libraries/script/config_dictionary.py", line 73, in __getattr__
    raise Fail("Configuration parameter '" + self.name + "' was not found in configurations dictionary!")
resource_management.core.exceptions.Fail: Configuration parameter 'kafka_broker_hosts' was not found in configurations dictionary!
 stdout:
2016-12-08 19:42:46,802 - The hadoop conf dir /usr/hdp/current/hadoop-client/conf exists, will call conf-select on it for version 2.5.0.0-1245
2016-12-08 19:42:46,806 - Checking if need to create versioned conf dir /etc/hadoop/2.5.0.0-1245/0
2016-12-08 19:42:46,811 - call[('ambari-python-wrap', u'/usr/bin/conf-select', 'create-conf-dir', '--package', 'hadoop', '--stack-version', '2.5.0.0-1245', '--conf-version', '0')] {'logoutput': False, 'sudo': True, 'quiet': False, 'stderr': -1}
2016-12-08 19:42:46,855 - call returned (1, '/etc/hadoop/2.5.0.0-1245/0 exist already', '')
2016-12-08 19:42:46,856 - checked_call[('ambari-python-wrap', u'/usr/bin/conf-select', 'set-conf-dir', '--package', 'hadoop', '--stack-version', '2.5.0.0-1245', '--conf-version', '0')] {'logoutput': False, 'sudo': True, 'quiet': False}
2016-12-08 19:42:46,897 - checked_call returned (0, '')
2016-12-08 19:42:46,898 - Ensuring that hadoop has the correct symlink structure
2016-12-08 19:42:46,898 - Using hadoop conf dir: /usr/hdp/current/hadoop-client/conf
2016-12-08 19:42:46,901 - Group['hadoop'] {}
2016-12-08 19:42:46,904 - Group['nifi'] {}
2016-12-08 19:42:46,904 - Group['users'] {}
2016-12-08 19:42:46,905 - User['hive'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'hadoop']}
2016-12-08 19:42:46,907 - User['zookeeper'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'hadoop']}
2016-12-08 19:42:46,908 - User['ambari-qa'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'users']}
2016-12-08 19:42:46,910 - User['tez'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'users']}
2016-12-08 19:42:46,911 - User['hdfs'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'hadoop']}
2016-12-08 19:42:46,912 - User['yarn'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'hadoop']}
2016-12-08 19:42:46,914 - User['nifi'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'hadoop']}
2016-12-08 19:42:46,915 - User['hcat'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'hadoop']}
2016-12-08 19:42:46,917 - User['mapred'] {'gid': 'hadoop', 'fetch_nonlocal_groups': True, 'groups': [u'hadoop']}
2016-12-08 19:42:46,919 - File['/var/lib/ambari-agent/tmp/changeUid.sh'] {'content': StaticFile('changeToSecureUid.sh'), 'mode': 0555}
2016-12-08 19:42:46,923 - Execute['/var/lib/ambari-agent/tmp/changeUid.sh ambari-qa /tmp/hadoop-ambari-qa,/tmp/hsperfdata_ambari-qa,/home/ambari-qa,/tmp/ambari-qa,/tmp/sqoop-ambari-qa'] {'not_if': '(test $(id -u ambari-qa) -gt 1000) || (false)'}
2016-12-08 19:42:46,936 - Skipping Execute['/var/lib/ambari-agent/tmp/changeUid.sh ambari-qa /tmp/hadoop-ambari-qa,/tmp/hsperfdata_ambari-qa,/home/ambari-qa,/tmp/ambari-qa,/tmp/sqoop-ambari-qa'] due to not_if
2016-12-08 19:42:46,937 - Group['hdfs'] {}
2016-12-08 19:42:46,938 - User['hdfs'] {'fetch_nonlocal_groups': True, 'groups': [u'hadoop', u'hdfs']}
2016-12-08 19:42:46,939 - FS Type: 
2016-12-08 19:42:46,939 - Directory['/etc/hadoop'] {'mode': 0755}
2016-12-08 19:42:46,970 - File['/usr/hdp/current/hadoop-client/conf/hadoop-env.sh'] {'content': InlineTemplate(...), 'owner': 'hdfs', 'group': 'hadoop'}
2016-12-08 19:42:46,972 - Directory['/var/lib/ambari-agent/tmp/hadoop_java_io_tmpdir'] {'owner': 'hdfs', 'group': 'hadoop', 'mode': 01777}
2016-12-08 19:42:46,999 - Initializing 2 repositories
2016-12-08 19:42:47,001 - Repository['HDP-2.5'] {'base_url': 'http://bigdata.persistent.co.in/HDP/HDP-2.5.0.0-centos7/', 'action': ['create'], 'components': [u'HDP', 'main'], 'repo_template': '[{{repo_id}}]\nname={{repo_id}}\n{% if mirror_list %}mirrorlist={{mirror_list}}{% else %}baseurl={{base_url}}{% endif %}\n\npath=/\nenabled=1\ngpgcheck=0', 'repo_file_name': 'HDP', 'mirror_list': None}
2016-12-08 19:42:47,016 - File['/etc/yum.repos.d/HDP.repo'] {'content': '[HDP-2.5]\nname=HDP-2.5\nbaseurl=http://bigdata.persistent.co.in/HDP/HDP-2.5.0.0-centos7/\n\npath=/\nenabled=1\ngpgcheck=0'}
2016-12-08 19:42:47,018 - Repository['HDP-UTILS-1.1.0.21'] {'base_url': 'http://bigdata.persistent.co.in/HDP/HDP-UTILS-1.1.0.21-centos7/', 'action': ['create'], 'components': [u'HDP-UTILS', 'main'], 'repo_template': '[{{repo_id}}]\nname={{repo_id}}\n{% if mirror_list %}mirrorlist={{mirror_list}}{% else %}baseurl={{base_url}}{% endif %}\n\npath=/\nenabled=1\ngpgcheck=0', 'repo_file_name': 'HDP-UTILS', 'mirror_list': None}
2016-12-08 19:42:47,025 - File['/etc/yum.repos.d/HDP-UTILS.repo'] {'content': '[HDP-UTILS-1.1.0.21]\nname=HDP-UTILS-1.1.0.21\nbaseurl=http://bigdata.persistent.co.in/HDP/HDP-UTILS-1.1.0.21-centos7/\n\npath=/\nenabled=1\ngpgcheck=0'}
2016-12-08 19:42:47,026 - Package['unzip'] {'retry_on_repo_unavailability': False, 'retry_count': 5}
2016-12-08 19:42:47,161 - Skipping installation of existing package unzip
2016-12-08 19:42:47,162 - Package['curl'] {'retry_on_repo_unavailability': False, 'retry_count': 5}
2016-12-08 19:42:47,181 - Skipping installation of existing package curl
2016-12-08 19:42:47,182 - Package['hdp-select'] {'retry_on_repo_unavailability': False, 'retry_count': 5}
2016-12-08 19:42:47,199 - Skipping installation of existing package hdp-select

Command failed after 1 tries

看答案

想知道你是否使用过 Ambari-Nifi服务 HDP？如果是这样，我认为项目被弃用，也许没有必要的更改与HDP 2.5完全兼容。但是，如果愿意尝试在您的末尾手动改变我会建议关注Nifi的Params.py中的这些行（从88号开始）

master_configs = config['clusterHostInfo']
nifi_master_hosts = master_configs['nifi_master_hosts']
kafka_broker_host = str(master_configs['kafka_broker_hosts'][0])

与您提到的补丁进行比较时，它看起来像他们从使用Config对象移动并调用默认方法来检索值（来自Ambari Patch）：

-kafka_broker_hosts = config['clusterHostInfo']['kafka_broker_hosts']
+kafka_broker_hosts = default('/clusterHostInfo/kafka_broker_hosts', [])

它可能值得尝试改变Nifi Params.py以做类似的事情：

master_configs = config['clusterHostInfo']
nifi_master_hosts = master_configs['nifi_master_hosts']
#new line
kafka_broker_hosts = default('/clusterHostInfo/kafka_broker_hosts', [])
#updated lne
kafka_broker_host = str(kafka_broker_hosts[0])

智能推荐

Apache Nifi

第1章NiFi入门介绍篇 1.1数据流挑战 1.1.1 数据系统演进传统的数据系统都是由单一的子系统组成，且各子系统之间没有数据交换的需求。近些年来随着移动互联网、物联网的快速发展，人类创造了史无前例的有价值的大数据。这些有价值的大数据往往不仅仅会存在于单一的子系统中，数据需要在复杂的各个系统之间进行传递和交换，让各种维度的数据能够聚集分析，从而产生出远远大于单个数据所产生的价值。下面让我们看一...

Apache Nifi

2.3.5 系统交互类处理器单元 ExecuteProcess ExecuteProcess处理器单元能够运行用户定义的操作系统命令，将处理完的标准输出内容写入flowfile中。该处理器是一个不需要输入的源处理器，它会输出产生一个新的FlowFile。如果需要提供输入源请使用下面介绍的executestreamcommand处理器单元。上面的图中表示Processor根据输入的指令和参数执行命...

Apache Nifi

2.3.4 属性提取类处理器单元 EvaluateJsonPath EvaluateJsonPath处理器单元根据用户定义的JSONpath表达式对FlowFile的JSON内容进行解析，将这些表达式所解析出来的内容替换FlowFile的内容或者将其更新到FlowFile的属性中，以便于后续的单元处理器的引用。上面的图中表示Processor将输入内容为JSON格式的FlowFile例如{&ldq...

Apache Nifi

2.3.3 数据接入类处理器单元 ConvertJSONToSQL ConvertJSONToSQL处理器单元可以将结构化的Json转换成INSERT或者UPDATE这样命令的SQL，配合PutSQL处理器单元可以直接根据这鞋命令将数据插入数据库中。上面的图中表示Processor根据输入的FlowFile的JSON内容，将JSON转化成Update的SQL语句。 ExecuteSQL Exec...

Apache Nifi

2.3.2 数据路由类和调制处理器单元 ControlRate ControlRate处理器单元用来控制数据流部分流量的速率。上面的图中的例子表示1分钟内只允许最多1000个FlowFile流过。 DetectDuplicate DetectDuplicate处理器单元用来依据用户定义的特征来监控和发现重复的FlowFile。通常这个处理器会搭配HashContent单元处理器来完成功能。上面...

猜你喜欢

Apache Nifi

2.3 NiFi Processor应用介绍对于NiFi的使用者来说，如果想要创建一个高效的数据流，那么就需要了解什么样的单元处理器才最适合这个数据流。NiFi拥有大量的可以用于各种业务场景的单元处理器可供使用者挑选和使用，这些单元处理器主要提供例如系统之间数据的传输，数据的路由，数据的转换、处理、分割和聚合等大类的功能。在每个NiFi的版本发布中都会有大量的新的处理器单元产生，这就导致本书中...

Apache NiFi

Apache Nifi 概念 1.1, NIFI简介 Apache NiFi 是一个易于使用，功能强大且可靠的系统，用于处理和分发数据。可以自动化管理系统间的数据流。它支持高度可配置的指示图的数据路由、转换和系统中介逻辑，支持从多种数据源动态拉取数据。NiFi原来是NSA的一个项目，目前已经代码开源，是Apache基金会的顶级项目之一。 NiFi是基于Java的，使用Maven支持包的构建管理。 ...

Apache NiFi深度扩展

介绍该高级文档旨在深入了解NiFi的实施和设计决策。它假设读者已经阅读了足够的其他文档来了解NiFi的基础知识。 FlowFiles是NiFi的核心，也是基于流程的设计。FlowFile是一种数据记录，由指向其内容（有效负载）的指针和支持内容的属性组成，该指针与一个或多个起源事件相关联。属性是用作FlowFile元数据的键/值对，例如FlowFile文件名。内容是文件的实际数据或有效负载。原产地...

Apache NiFi 概述

Apache NiFi概述 Apache NiFi团队[email protected] 什么是Apache NiFi？简单地说，NiFi是为了自动化系统之间的数据流而构建的。虽然术语“数据流”在各种环境中使用，但我们在此处使用它来表示系统之间自动化和管理的信息流。这个问题空间一直存在，因为企业有多个系统，其中一些系统创建数据，一些系统消耗数据。已经讨论...

Apache NiFi 入门指南

本指南使用于谁？本指南适用于从未使用过，在NiFi中有限度接触或仅完成特定任务的用户。本指南不是详尽的说明手册或参考指南。“ 用户指南”提供了大量信息，旨在提供更加详尽的资源，并且作为参考指南非常有用。相比之下，本指南旨在为用户提供所需的信息，以便了解如何使用NiFi，以便快速轻松地构建强大而灵活的数据流。一些因为本指南中的某些信息仅适用于初次使用的用户，而其...