华为云用户手册

前提条件 创建或获取该任务中创建Loader作业的业务用户和密码。 确保用户已授权访问作业执行时操作的HDFS/OBS目录和数据。 获取SFTP服务器使用的用户和密码，且该用户具备SFTP服务器数据导出目录的写入权限。 检查磁盘空间，确保没有出现告警且余量满足导入、导出数据的大小。 使用Loader从HDFS/OBS导出数据时，确保HDFS/OBS数据源的输入路径目录名、输入路径的子目录名及子文件名不能包含特殊字符/\"':;,中的任意字符。 如果设置的任务需要使用指定YARN队列功能，该用户需要已授权有相关YARN队列的权限。 设置任务的用户需要获取该任务的执行权限，并获取该任务对应的连接的使用权限。

前提条件 创建或获取该任务中创建Loader作业的业务用户和密码。 确保用户已授权访问作业执行时操作的目录、HBase表和数据。 获取外部数据源（SFTP服务器或关系型数据库）使用的用户和密码。 检查磁盘空间，确保没有出现告警且余量满足导入、导出数据的大小。 使用Loader从HDFS/OBS导出数据时，确保HDFS/OBS数据源的输入路径目录名、输入路径的子目录名及子文件名不能包含特殊字符/\"':;,中的任意字符。 如果设置的任务需要使用指定Yarn队列功能，该用户需要已授权有相关Yarn队列的权限。 设置任务的用户需要获取该任务的执行权限，并获取该任务对应的连接的使用权限。

前提条件 创建或获取该任务中创建Loader作业的业务用户和密码。 确保用户已授权访问作业执行时操作的HDFS/OBS目录和数据。 确保用户已授权访问作业执行时操作的HBase表或phoenix表。 检查磁盘空间，确保没有出现告警且余量满足导入、导出数据的大小。 使用Loader从HDFS/OBS导入数据时，确保HDFS/OBS输入路径目录名、输入路径的子目录名及子文件名不能包含特殊字符/\"':;,中的任意字符。 如果设置的作业需要使用指定YARN队列功能，该用户需要已授权有相关YARN队列的权限。 设置任务的用户需要获取该任务的执行权限，并获取该任务对应的连接的使用权限。

前提条件 创建或获取该任务中创建Loader作业的业务用户和密码。 获取FTP服务器使用的用户和密码，且该用户具备FTP服务器上源文件的读取权限。如果源文件在导入后文件名要增加后缀，则该用户还需具备源文件的写入权限。 检查磁盘空间，确保没有出现告警且余量满足导入、导出数据的大小。 使用Loader从FTP服务器导入数据时，确保FTP服务器输入路径目录名、输入路径的子目录名及子文件名不能包含特殊字符/\"':;,中的任意字符。 如果设置的作业需要使用指定YARN队列功能，该用户需要已授权有相关YARN队列的权限。 设置任务的用户需要获取该任务的执行权限，并获取该任务对应的连接的使用权限。

前提条件 创建或获取该任务中创建Loader作业的业务用户和密码。 确保用户已授权访问作业中指定的Hive表的权限。 获取SFTP服务器使用的用户和密码，且该用户具备SFTP服务器上源文件的读取权限。如果源文件在导入后文件名要增加后缀，则该用户还需具备源文件的写入权限。 检查磁盘空间，确保没有出现告警且余量满足导入、导出数据的大小。 使用Loader从SFTP服务器导入数据时，确保SFTP服务器输入路径目录名、输入路径的子目录名及子文件名不能包含特殊字符/\"':;,中的任意字符。 如果设置的作业需要使用指定YARN队列功能，该用户需要已授权有相关YARN队列的权限。 设置任务的用户需要获取该任务的执行权限，并获取该任务对应的连接的使用权限。

前提条件 创建或获取该任务中创建Loader作业的业务用户和密码。 确保用户已授权访问作业执行时操作的HBase表或phoenix表。 获取SFTP服务器使用的用户和密码，且该用户具备SFTP服务器上源文件的读取权限。如果源文件在导入后文件名要增加后缀，则该用户还需具备源文件的写入权限。 检查磁盘空间，确保没有出现告警且余量满足导入、导出数据的大小。 使用Loader从SFTP服务器导入数据时，确保SFTP服务器输入路径目录名、输入路径的子目录名及子文件名不能包含特殊字符/\"':;,中的任意字符。 如果设置的作业需要使用指定YARN队列功能，该用户需要已授权有相关YARN队列的权限。 设置任务的用户需要获取该任务的执行权限，并获取该任务对应的连接的使用权限。

Loader数据导入简介 Loader是实现 MRS 与外部数据源如关系型数据库、SFTP服务器、FTP服务器之间交换数据和文件的ETL工具，支持将数据或文件从关系型数据库或文件系统导入到MRS系统中。 Loader支持如下数据导入方式： 从关系型数据库导入数据到HDFS/OBS 从关系型数据库导入数据到HBase 从关系型数据库导入数据到Phoenix表 从关系型数据库导入数据到Hive表 从SFTP服务器导入数据到HDFS/OBS 从SFTP服务器导入数据到HBase 从SFTP服务器导入数据到Phoenix表 从SFTP服务器导入数据到Hive表 从FTP服务器导入数据到HDFS/OBS 从FTP服务器导入数据到HBase 从FTP服务器导入数据到Phoenix表 从FTP服务器导入数据到Hive表 从同一集群内HDFS/OBS导入数据到HBase MRS与外部数据源交换数据和文件时需要连接数据源。系统提供以下连接器，用于配置不同类型数据源的连接参数： generic-jdbc-connector：关系型数据库连接器。 ftp-connector：FTP数据源连接器。 hdfs-connector：HDFS数据源连接器。 oracle-connector：Oracle数据库专用连接器，使用row_id作为分区列，相对generic-jdbc-connector来说，Map任务分区更均匀，并且不依赖分区列是否有创建索引。 mysql-fastpath-connector：MYSQL数据库专用连接器，使用MYSQL的mysqldump和mysqlimport工具进行数据的导入导出，相对generic-jdbc-connector来说，导入导出速度更快。 sftp-connector：SFTP数据源连接器。 oracle-partition-connector：支持Oracle分区特性的连接器，专门对Oracle分区表的导入导出进行优化。 使用FTP数据源连接器时不加密数据，可能存在安全风险，建议使用SFTP数据源连接器。 建议将SFTP服务器、FTP服务器和数据库服务器与Loader部署在独立的子网中，以保障数据安全地导入。 与关系数据库连接时，可以选择通用数据库连接器（generic-jdbc-connector）或者专用数据库连接器（oracle-connector、oracle-partition-connector、mysql-fastpath-connector），专用数据库连接器特别针对具体数据库类型进行优化，相对通用数据库连接器来说，导出、导入速度更快。 使用mysql-fastpath-connector时，要求在NodeManager节点上有MySQL的mysqldump和mysqlimport命令，并且此两个命令所属MySQL客户端版本与MySQL服务器版本兼容，如果没有这两个命令或版本不兼容，请参考http://dev.mysql.com/doc/refman/5.7/en/linux-installation-rpm.html，安装MySQL client applications and tools。 使用oracle-connector时，要求给连接用户赋予如下系统表或者视图的select权限： dba_tab_partitions、dba_constraints、dba_tables 、dba_segments 、v$version、dba_objects、v$instance、SYS_CONTEXT函数、dba_extents、 dba_tab_subpartitions。 使用oracle-partition-connector时，要求给连接用户赋予如下系统表的select权限：dba_objects、dba_extents。

Loader数据导出简介 Loader支持将数据或者文件从MRS系统中导出到关系型数据库或文件系统中，Loader支持如下数据导出方式： 从HDFS/OBS中导出数据到SFTP服务器 从HDFS/OBS中导出数据到关系型数据库 从HBase中导出数据到SFTP服务器 从HBase中导出数据到关系型数据库 从Phoenix表导出数据到SFTP服务器 从Phoenix表导出数据到关系型数据库 从Hive中导出数据到SFTP服务器 从Hive中导出数据到关系数据库 从同一集群内HBase导出数据到HDFS/OBS

如何开发Flume第三方插件 安装Flume客户端，如安装目录为“/opt/flumeclient”。 将自主研发的代码打成jar包。 建立插件目录布局。 进入“Flume客户端安装目录/fusionInsight-flume-*/plugins.d”路径下，使用以下命令建立目录，可根据实际业务进行命名，无固定名称： cd /opt/flumeclient/fusioninsight-flume-1.9.0/plugins.d mkdir thirdPlugin cd thirdPlugin mkdir lib libext native 显示结果如下： 将第三方jar包放入“Flume客户端安装目录/fusionInsight-flume-*/plugins.d/thirdPlugin/lib”路径下，如果该jar包依赖其他jar包，则将所依赖的jar包放入“Flume客户端安装目录/fusionInsight-flume-*/plugins.d/thirdPlugin/libext”文件夹中，“Flume客户端安装目录/fusionInsight-flume-*/plugins.d/thirdPlugin/native”放置本地库文件。 配置“Flume客户端安装目录/fusionInsight-flume-*/conf/properties.properties”文件。 具体properties.properties参数配置方法，参考配置Flume非加密传输数据采集任务和配置Flume加密传输数据采集任务对应典型场景中properties.properties文件参数列表的说明。 父主题： Flume常见问题

日志级别 Flume提供了如表2所示的日志级别。 运行日志的级别优先级从高到低分别是FATAL、ERROR、WARN、INFO、DEBUG，程序会打印高于或等于所设置级别的日志，设置的日志等级越高，打印出来的日志就越少。 表2 日志级别 日志类型 级别 描述 运行日志 FATAL FATAL表示系统运行的致命错误信息。 ERROR ERROR表示系统运行的错误信息。 WARN WARN表示当前事件处理存在异常信息。 INFO INFO表示记录系统及各事件正常运行状态信息。 DEBUG DEBUG表示记录系统及系统的调试信息。 如果您需要修改日志级别，请执行如下操作： 请参考修改集群服务配置参数，进入Flume的“全部配置”页面。 左边菜单栏中选择所需修改的角色所对应的日志菜单。 选择所需修改的日志级别。 保存配置，在弹出窗口中单击“确定”使配置生效。 配置完成后即生效，不需要重启服务。

注意事项 Flume可靠性保障措施有哪些？ Source&Channel、Channel&Sink之间的事务机制。 Sink Processor支持配置failover、load_blance机制，例如负载均衡示例如下。 server.sinkgroups=g1 server.sinkgroups.g1.sinks=k1 k2 server.sinkgroups.g1.processor.type=load_balance server.sinkgroups.g1.processor.backoff=true server.sinkgroups.g1.processor.selector=random Flume多agent聚合级联时的注意事项？ 级联时需要使用Avro或者Thrift协议进行级联。 聚合端存在多个节点时，连接配置尽量配置均衡，不要聚合到单节点上。

常用Sink配置 HDFS Sink HDFS Sink将数据写入Hadoop分布式文件系统（HDFS）。常用配置如下表所示： 表16 HDFS Sink常用配置 参数 默认值 描述 channel - 与之相连的channel。 type hdfs hdfs sink的类型，必须设置为hdfs。 hdfs.path - HDFS上数据存储路径，必须以“hdfs://hacluster/”开头。 monTime 0（不开启） 线程监控阈值，更新时间超过阈值后，重新启动该Sink，单位：秒。 hdfs.inUseSuffix .tmp 正在写入的hdfs文件后缀。 hdfs.rollInterval 30 按时间滚动文件，单位：秒，同时需将“hdfs.fileCloseByEndEvent”设置为“false”。 hdfs.rollSize 1024 按大小滚动文件，单位：bytes，同时需将“hdfs.fileCloseByEndEvent”设置为“false”。 hdfs.rollCount 10 按Event个数滚动文件，同时需将“hdfs.fileCloseByEndEvent”设置为“false”。 说明： 参数“rollInterval”、“rollSize”和“rollCount”可同时配置，三个参数采取优先原则，哪个参数值先满足，优先按照哪个参数进行压缩。 hdfs.idleTimeout 0 自动关闭空闲文件超时时间，单位：秒。 hdfs.batchSize 1000 批次写入HDFS的Event个数。 hdfs.kerberosPrincipal - 认证HDFS的Kerberos principal，普通模式集群不配置，安全模式集群必须配置。 hdfs.kerberosKeytab - 认证HDFS的Kerberos keytab，普通模式集群不配置，安全模式集群中，用户必须对jaas.cof文件中的keyTab路径有访问权限。 hdfs.fileCloseByEndEvent true 收到源文件的最后一个Event时是否关闭hdfs文件。 hdfs.batchCallTimeout - 批次写入HDFS超时控制时间，单位：毫秒。 当不配置此参数时，对每个Event写入HDFS进行超时控制。当“hdfs.batchSize”大于0时，配置此参数可以提升写入HDFS性能。 说明： “hdfs.batchCallTimeout”设置多长时间需要考虑“hdfs.batchSize”的大小，“hdfs.batchSize”越大，“hdfs.batchCallTimeout”也要调整更长时间，设置过短时间容易导致写HDFS失败。 serializer.appendNewline true 将一个Event写入HDFS后是否追加换行符（'\n'），如果追加该换行符，该换行符所占用的数据量指标不会被HDFS Sink统计。 hdfs.filePrefix over_%{basename} 数据写入hdfs后文件名的前缀。 hdfs.fileSuffix - 数据写入hdfs后文件名的后缀。 hdfs.inUsePrefix - 正在写入的hdfs文件前缀。 hdfs.fileType DataStream hdfs文件格式，包括“SequenceFile”、“DataStream”以及“CompressedStream”。 说明： “SequenceFile”和“DataStream”不压缩输出文件，不能设置参数“codeC”，“CompressedStream”压缩输出文件，必须设置“codeC”参数值配合使用。 hdfs.codeC - 文件压缩格式，包括gzip、bzip2、lzo、lzop、snappy。 hdfs.maxOpenFiles 5000 最大允许打开的hdfs文件数，当打开的文件数达到该值时，最早打开的文件将会被关闭。 hdfs.writeFormat Writable 文件写入格式，“Writable”或者“Text”。 hdfs.callTimeout 10000 写入HDFS超时控制时间，单位：毫秒。 hdfs.threadsPoolSize - 每个HDFS sink用于HDFS io操作的线程数。 hdfs.rollTimerPoolSize - 每个HDFS sink用于调度定时文件滚动的线程数。 hdfs.round false 时间戳是否四舍五入。如果设置为true，则会影响所有基于时间的转义序列（％t除外）。 hdfs.roundUnit second 时间戳四舍五入单位，可选为“second”、“minute”或“hour”，分别对应为秒、分钟和小时。 hdfs.useLocalTimeStamp true 是否启用本地时间戳，建议设置为“true”。 hdfs.closeTries 0 hdfs sink尝试关闭重命名文件的最大次数。默认为0表示sink会一直尝试重命名，直至重命名成功。 hdfs.retryInterval 180 尝试关闭hdfs文件的时间间隔，单位：秒。 说明： 每个关闭请求都会有多个RPC往返Namenode，因此设置的太低可能导致Namenode超负荷。如果设置0，如果第一次尝试失败的话，该Sink将不会尝试关闭文件，并且把文件打开，或者用“.tmp”作为扩展名。 hdfs.failcount 10 数据写入hdfs失败的次数。该参数作为sink写入hdfs失败次数的阈值，当超过该阈值后上报数据传输异常告警。 Avro Sink Avro Sink把events转化为Avro events并发送到配置的主机的监测端口。常用配置如下表所示： 表17 Avro Sink常用配置 参数 默认值 描述 channel - 与之相连的channel。 type - avro sink的类型，必须设置为avro。 hostname - 绑定的主机名/IP。 port - 监测端口，该端口需未被占用。 batch-size 1000 批次发送的Event个数。 client.type DEFAULT 客户端实例类型，根据所配置的模型实际使用到的通信协议设置。该值可选值包括： DEFAULT，返回AvroRPC类型的客户端实例。 OTHER，返回NULL。 THRIFT，返回Thrift RPC类型的客户端实例。 DEFAULT_LOADBALANCING， 返回LoadBalancing RPC 客户端实例。 DEFAULT_FAILOVER， 返回Failover RPC 客户端实例。 ssl false 是否使用SSL加密。设置为true时还必须指定“密钥(keystore)”和“密钥存储密码(keystore-password)”。 truststore-type JKS Java信任库类型，“JKS”或“PK CS 12”。 说明： JKS的密钥库和私钥采用不同的密码进行保护，而PKCS12的密钥库和私钥采用相同密码进行保护。 truststore - Java信任库文件。 truststore-password - Java信任库密码。 keystore-type JKS ssl启用后密钥存储类型。 keystore - ssl启用后密钥存储文件路径，开启ssl后，该参数必填。 keystore-password - ssl启用后密钥存储密码，开启ssl后，该参数必填。 connect-timeout 20000 第一次连接的超时时间，单位：毫秒。 request-timeout 20000 第一次请求后一次请求的最大超时时间，单位：毫秒。 reset-connection-interval 0 一次断开连接后，等待多少时间后进行重新连接，单位：秒。默认为0表示不断尝试。 compression-type none 批数据压缩类型，“none”或“deflate”，“none”表示不压缩，“deflate”表示压缩。该值必须与AvroSource的compression-type匹配。 compression-level 6 批数据压缩级别（1-9），数值越高，压缩率越高。 exclude-protocols SSLv3 排除的协议列表，用空格分开。默认排除SSLv3协议。 HBase Sink HBase Sink将数据写入到HBase中。常用配置如下表所示： 表18 HBase Sink常用配置 参数 默认值 描述 channel - 与之相连的channel。 type - hbase sink的类型，必须设置为hbase。 table - HBase表名称。 columnFamily - HBase列族。 monTime 0（不开启） 线程监控阈值，更新时间超过阈值后，重新启动该Sink，单位：秒。 batchSize 1000 批次写入HBase的Event个数。 kerberosPrincipal - 认证HBase的Kerberos principal，普通模式集群不配置，安全模式集群必须配置。 kerberosKeytab - 认证HBase的Kerberos keytab，普通模式集群不配置，安全模式集群中，flume运行用户必须对jaas.cof文件中的keyTab路径有访问权限。 coalesceIncrements true 是否在同一个处理批次中，合并对同一个hbase cell多个操作。设置为true有利于提高性能。 Kafka Sink Kafka Sink将数据写入到Kafka中。常用配置如下表所示： 表19 Kafka Sink常用配置 参数 默认值 描述 channel - 与之相连的channel。 type - kafka sink的类型，必须设置为org.apache.flume.sink.kafka.KafkaSink。 kafka.bootstrap.servers - Kafka 的bootstrap 地址端口列表。如果集群安装有kafka并且配置已经同步，服务端可以不配置此项，默认值为Kafka集群中所有的broker列表，客户端必须配置该项，多个用逗号分隔。端口和安全协议的匹配规则必须为：21007匹配安全模式（SASL_PLAINTEXT），9092匹配普通模式（PLAINTEXT）。 monTime 0（不开启） 线程监控阈值，更新时间超过阈值后，重新启动该Sink，单位：秒。 kafka.producer.acks 1 必须收到多少个replicas的确认信息才认为写入成功。0表示不需要接收确认信息，1表示只等待leader的确认信息。-1表示等待所有的relicas的确认信息。设置为-1，在某些leader失败的场景中可以避免数据丢失。 kafka.topic - 数据写入的topic，必须填写。 allowTopicOverride false 是否将Event Header中保存的topic替换kafka.topic中配置的topic。 flumeBatchSize 1000 批次写入Kafka的Event个数。 kafka.security.protocol SASL_PLAINTEXT Kafka安全协议，普通模式集群下须配置为“PLAINTEXT”。端口和安全协议的匹配规则必须为：21007匹配安全模式（SASL_PLAINTEXT），9092匹配普通模式（PLAINTEXT）。 ignoreLongMessage false 是否丢弃超大消息的开关。 messageMaxLength 1000012 Flume写入Kafka的消息的最大长度。 defaultPartitionId - 用于指定channel中的events被传输到哪一个Kafka partition ID ，此值会被partitionIdHeader覆盖。默认情况下，如果此参数不设置，会由Kafka Producer's partitioner 进行events分发(可以通过指定key或者kafka.partitioner.class自定义的partitioner)。 partitionIdHeader - 设置时，对应的Sink 将从Event 的Header中获取使用此属性的值命名的字段的值，并将消息发送到主题的指定分区。 如果该值无对应的有效分区，则会发生EventDeliveryException。 如果Header 值已经存在，则此设置将覆盖参数defaultPartitionId。 Other Kafka Producer Properties - 其他Kafka配置，可以接受任意Kafka支持的生产配置，配置需要加前缀 .kafka。 Thrift Sink Thrift Sink把events转化为Thrift events并发送到配置的主机的监测端口。常用配置如下表所示： 表20 Thrift Sink常用配置 参数 默认值 描述 channel - 与之相连的channel。 type thrift thrift sink的类型，必须设置为thrift。 hostname - 绑定的主机名/IP。 port - 监测端口，该端口需未被占用。 batch-size 1000 批次发送的Event个数。 connect-timeout 20000 第一次连接的超时时间，单位：毫秒。 request-timeout 20000 第一次请求后一次请求的最大超时时间，单位：毫秒。 kerberos false 是否启用Kerberos认证。 client-keytab - 客户端使用的keytab文件地址，flume运行用户必须对认证文件具有访问权限。 client-principal - 客户端使用的安全用户的Principal。 server-principal - 服务端使用的安全用户的Principal。 compression-type none Flume发送数据的压缩类型，“none”或“deflate”，“none”表示不压缩，“deflate”表示压缩。 maxConnections 5 Flume发送数据时的最大连接池大小。 ssl false 是否使用SSL加密。 truststore-type JKS Java信任库类型。 truststore - Java信任库文件。 truststore-password - Java信任库密码。 reset-connection-interval 0 一次断开连接后，等待多少时间后进行重新连接，单位：秒。默认为0表示不断尝试。

常用Channel配置 Memory Channel Memory Channel使用内存作为缓存区，Events存放在内存队列中。常用配置如下表所示： 表12 Memory Channel常用配置 参数 默认值 描述 type - memory channel的类型，必须设置为memory。 capacity 10000 缓存在channel中的最大Event数。 transactionCapacity 1000 每次存取的最大Event数。 说明： 此参数值需要大于source和sink的batchSize。 事务缓存容量必须小于或等于Channel缓存容量。 channelfullcount 10 channel full次数，达到该次数后发送告警。 keep-alive 3 当事务缓存或Channel缓存满时，Put、Take线程等待时间。单位：秒。 byteCapacity JVM最大内存的80% channel中最多能容纳所有event body的总字节数，默认是 JVM最大可用内存（-Xmx ）的80%，单位：bytes。 byteCapacityBufferPercentage 20 channel中字节容量百分比（%）。 File Channel File Channel使用本地磁盘作为缓存区，Events存放在设置的dataDirs配置项文件夹中。常用配置如下表所示： 表13 File Channel常用配置 参数 默认值 描述 type - file channel的类型，必须设置为file。 checkpointDir ${BIGDATA_DATA_HOME}/hadoop/data1~N/flume/checkpoint 说明： 此路径随自定义数据路径变更。 检查点存放路径。 dataDirs ${BIGDATA_DATA_HOME}/hadoop/data1~N/flume/data 说明： 此路径随自定义数据路径变更。 数据缓存路径，设置多个路径可提升性能，中间用逗号分开。 maxFileSize 2146435071 单个缓存文件的最大值，单位：bytes。 minimumRequiredSpace 524288000 缓冲区空闲空间最小值，单位：bytes。 capacity 1000000 缓存在channel中的最大Event数。 transactionCapacity 10000 每次存取的最大Event数。 说明： 此参数值需要大于source和sink的batchSize。 事务缓存容量必须小于或等于Channel缓存容量。 channelfullcount 10 channel full次数，达到该次数后发送告警。 useDualCheckpoints false 是否备份检查点。设置为“true”时，必须设置backupCheckpointDir的参数值。 backupCheckpointDir - 备份检查点路径。 checkpointInterval 30000 检查点间隔时间，单位：秒。 keep-alive 3 当事务缓存或Channel缓存满时，Put、Take线程等待时间。单位：秒。 use-log-replay-v1 false 是否启用旧的回复逻辑。 use-fast-replay false 是否使用队列回复。 checkpointOnClose true channel关闭时是否创建检查点。 Memory File Channel Memory File Channel同时使用内存和本地磁盘作为缓存区，消息可持久化，性能优于File Channel，接近Memory Channel的性能。此Channel目前处于试验阶段，可靠性不够高，不建议在生产环境使用。常用配置如下表所示： 表14 Memory File Channel常用配置 参数 默认值 描述 type org.apache.flume.channel.MemoryFileChannel memory file channel的类型，必须设置为“org.apache.flume.channel.MemoryFileChannel”。 capacity 50000 Channel缓存容量：缓存在Channel中的最大Event数。 transactionCapacity 5000 事务缓存容量：一次事务能处理的最大Event数。 说明： 此参数值需要大于source和sink的batchSize。 事务缓存容量必须小于或等于Channel缓存容量。 subqueueByteCapacity 20971520 每个subqueue最多保存多少byte的Event，单位：byte。 Memory File Channel采用queue和subqueue两级缓存，event保存在subqueue，subqueue保存在queue。 subqueue能保存多少event，由“subqueueCapacity”和“subqueueInterval”两个参数决定，“subqueueCapacity”限制subqueue内的Event总容量，“subqueueInterval”限制subqueue保存Event的时长，只有subqueue达到“subqueueCapacity”或“subqueueInterval”上限时，subqueue内的Event才会发往目的地。 说明： “subqueueByteCapacity”必须大于一个batchsize内的Event总容量。 subqueueInterval 2000 每个subqueue最多保存一段多长时间的Event，单位：毫秒。 keep-alive 3 当事务缓存或Channel缓存满时，Put、Take线程等待时间。 单位：秒。 dataDir - 缓存本地文件存储目录。 byteCapacity JVM最大内存的80% Channel缓存容量。 单位：bytes。 compression-type None 消息压缩格式：“none”或“deflate”。“none”表示不压缩，“deflate”表示压缩。 channelfullcount 10 channel full次数，达到该次数后发送告警。 Memory File Channel配置样例： server.channels.c1.type = org.apache.flume.channel.MemoryFileChannel server.channels.c1.dataDir = /opt/flume/mfdata server.channels.c1.subqueueByteCapacity = 20971520 server.channels.c1.subqueueInterval=2000 server.channels.c1.capacity = 500000 server.channels.c1.transactionCapacity = 40000 Kafka Channel Kafka Channel使用Kafka集群缓存数据，Kafka提供高可用、多副本，以防Flume或Kafka Broker崩溃，Channel中的数据会立即被Sink消费。 表15 Kafka channel 常用配置 Parameter Default Value Description type - kafka channel的类型，必须设置为 “org.apache.flume.channel.kafka.KafkaChannel”。 kafka.bootstrap.servers - Kafka的bootstrap地址端口列表。 如果集群已安装Kafka并且配置已经同步，则服务端可以不配置此项，默认值为Kafka集群中所有的broker列表。客户端必须配置该项，多个值用逗号分隔。端口和安全协议的匹配规则必须为：21007匹配安全模式（SASL_PLAINTEXT），9092匹配普通模式（PLAINTEXT）。 kafka.topic flume-channel channel用来缓存数据的topic。 kafka.consumer.group.id flume 从kafka中获取数据的组标识，此参数不能为空。 parseAsFlumeEvent true 是否解析为Flume event。 migrateZookeeperOffsets true 当Kafka没有存储offset时，是否从ZooKeeper中查找，并提交到Kafka。 kafka.consumer.auto.offset.reset latest 当没有offset记录时从什么位置消费，可选为“earliest”、“latest”或“none”。“earliest”表示将offset重置为初始点，“latest”表示将offset置为最新位置点，“none”表示如果没有offset则发生异常。 kafka.producer.security.protocol SASL_PLAINTEXT Kafka生产安全协议。端口和安全协议的匹配规则必须为：21007匹配安全模式（SASL_PLAINTEXT），9092匹配普通模式（PLAINTEXT）。 说明： 如果该参数没有显示，请单击弹窗左下角的"+"显示全部参数。 kafka.consumer.security.protocol SASL_PLAINTEXT 同上，但用于消费。端口和安全协议的匹配规则必须为：21007匹配安全模式（SASL_PLAINTEXT），9092匹配普通模式（PLAINTEXT）。 pollTimeout 500 consumer调用poll()函数能接受的最大超时时间，单位：毫秒。 ignoreLongMessage false 是否丢弃超大消息。 messageMaxLength 1000012 Flume写入Kafka的消息的最大长度。

Channel Selector Channel Selector可以允许一个Source对接多个Channel，通过选择不同的Selector类型来将Source的数据进行分流或者复制，目前Flume提供的Channel Selector有两种：Replicating和Multiplexing。 Replicating：表示Source的数据同步发送给所有Channel。 Multiplexing：表示根据Event中的Header的指定字段的值来进行判断，从而选择相应的Channel进行发送，从而起到根据业务类型进行分流的目的。 Replicating配置样例： client.sources = kafkasource client.channels = channel1 channel2 client.sources.kafkasource.type = org.apache.flume.source.kafka.KafkaSource client.sources.kafkasource.kafka.topics = topic1,topic2 client.sources.kafkasource.kafka.consumer.group.id = flume client.sources.kafkasource.kafka.bootstrap.servers = 10.69.112.108:21007 client.sources.kafkasource.kafka.security.protocol = SASL_PLAINTEXT client.sources.kafkasource.batchDurationMillis = 1000 client.sources.kafkasource.batchSize = 800 client.sources.kafkasource.channels = channel1 channel2 client.sources.kafkasource.selector.type = replicating client.sources.kafkasource.selector.optional = channel2 表1 Replicating配置样例参数说明 选项名称 默认值 描述 Selector.type replicating Selector类型，应配置为replicating Selector.optional - 可选Channel，可以配置为列表 Multiplexing配置样例： client.sources = kafkasource client.channels = channel1 channel2 client.sources.kafkasource.type = org.apache.flume.source.kafka.KafkaSource client.sources.kafkasource.kafka.topics = topic1,topic2 client.sources.kafkasource.kafka.consumer.group.id = flume client.sources.kafkasource.kafka.bootstrap.servers = 10.69.112.108:21007 client.sources.kafkasource.kafka.security.protocol = SASL_PLAINTEXT client.sources.kafkasource.batchDurationMillis = 1000 client.sources.kafkasource.batchSize = 800 client.sources.kafkasource.channels = channel1 channel2 client.sources.kafkasource.selector.type = multiplexing client.sources.kafkasource.selector.header = myheader client.sources.kafkasource.selector.mapping.topic1 = channel1 client.sources.kafkasource.selector.mapping.topic2 = channel2 client.sources.kafkasource.selector.default = channel1 表2 Multiplexing配置样例参数说明 选项名称 默认值 描述 Selector.type replicating Selector类型，应配置为multiplexing Selector.header Flume.selector.header - Multiplexing类型的Selector的样例中，选择Event中Header名称为topic的字段来进行判断，当Header中topic字段的值为topic1时，向channel1发送该Event，当Header中topic字段的值为topic2时，向channel2发送该Event。 这种Selector需要借助Source中Event的特定Header来进行Channel的选择，需要根据业务场景选择合理的Header来进行数据分流。

Interceptors Flume的拦截器（Interceptor）支持在数据传输过程中修改或丢弃传输的基本单元Event。用户可以通过在配置中指定Flume内建拦截器的类名列表，也可以开发自定义的拦截器来实现Event的修改或丢弃。Flume内建支持的拦截器如下表所示，本章节会选取一个较为复杂的作为示例。其余的用户可以根据需要自行配置使用。 拦截器用在Flume的Source、Channel之间，大部分的Source都带有Interceptor参数。用户可以依据需要配置。 Flume支持一个Source配置多个拦截器，各拦截器名称用空格分开。 指定拦截器的顺序就是它们被调用的顺序。 使用拦截器在Header中插入的内容，都可以在Sink中读取并使用。 表5 Flume内建支持的拦截器类型 拦截器类型 简要描述 Timestamp Interceptor 该拦截器会在Event的Header中插入一个时间戳。 Host Interceptor 该拦截器会在Event的Header中插入当前Agent所在节点的IP或主机名。 Remove Header Interceptor 该拦截器会依据Header中包含的符合正则匹配的字符串，丢弃掉对应的Event。 UUID Interceptor 该拦截器会为每个Event的Header生成一个UUID字符串。 Search and Replace Interceptor 该拦截器基于Java正则表达式提供简单的基于字符串的搜索和替换功能。与Java Matcher.replaceAll() 的规则相同。 Regex Filtering Interceptor 该拦截器通过将Event的Body体解释为文本文件，与配置的正则表达式进行匹配来选择性的过滤Event。提供的正则表达式可用于排除或包含事件。 Regex Extractor Interceptor 该拦截器使用正则表达式抽取原始events中的内容，并将该内容加入events的header中。 下面以Regex Filtering Interceptor 为例说明Interceptor使用（其余的可参考官网配置）： 表6 Regex Filtering Interceptor配置参数说明 选项名称 默认值 描述 type - 组件类型名称，必须写为regex_filter。 regex - 用于匹配事件的正则表达式。 excludeEvents false 默认收集匹配到的Event。设置为true，则会删除匹配的Event，保留不匹配的。 配置示例（为了方便观察，此模型使用了netcat tcp作为Source源，logger作为Sink）。配置好如下参数后，在Linux的配置的主机节点上执行Linux命令“telnet 主机名或IP 44444”，并任意敲入符合正则和不符合正则的字符串。会在日志中观察到，只有匹配到的字符串被传输了。 #define the source、channel、sink server.sources = r1 server.channels = c1 server.sinks = k1 #config the source server.sources.r1.type = netcat server.sources.r1.bind = ${主机IP} server.sources.r1.port = 44444 server.sources.r1.interceptors= i1 server.sources.r1.interceptors.i1.type= regex_filter server.sources.r1.interceptors.i1.regex= (flume)|(myflume) server.sources.r1.interceptors.i1.excludeEvents= false server.sources.r1.channels = c1 #config the channel server.channels.c1.type = memory server.channels.c1.capacity = 1000 server.channels.c1.transactionCapacity = 100 #config the sink server.sinks.k1.type = logger server.sinks.k1.channel = c1

业务模型配置指导 本任务旨在提供Flume常用模块的性能差异，用于指导用户进行合理的Flume业务配置，避免出现前端Source和后端Sink性能不匹配进而导致整体业务性能不达标的场景。 本任务只针对于单通道的场景进行比较说明。 Flume业务配置及模块选择过程中，一般要求Sink的极限吞吐量需要大于Source的极限吞吐量，否则在极限负载的场景下，Source往Channel的写入速度大于Sink从Channel取出的速度，从而导致Channel频繁被写满，进而影响性能表现。 Avro Source和Avro Sink一般都是成对出现，用于多个Flume Agent间进行数据中转，因此一般场景下Avro Source和Avro Sink都不会成为性能瓶颈。

模块间性能 根据模块间性能对比，可以看到对于前端是SpoolDir Source的场景下，Kafka Sink和HDFS Sink都能满足吞吐量要求，但是HBase Sink由于自身写入性能较低的原因，会成为性能瓶颈，会导致数据都积压在Channel中。但是如果有必须使用HBase Sink或者其他性能容易成为瓶颈的Sink的场景时，可以选择使用Channel Selector或者Sink Group来满足性能要求。

参数调优 修改服务配置参数，请参考修改集群服务配置参数。调优参数请参考表1。 表1 调优参数 配置参数 缺省值 调优场景 num.recovery.threads.per.data.dir 10 在Kafka启动过程中，数据量较大情况下，可调大此参数，可以提升启动速度。 background.threads 10 Broker后台任务处理的线程数目。数据量较大的情况下，可适当调大此参数，以提升Broker处理能力。 num.replica.fetchers 1 副本向Leader请求同步数据的线程数，增大这个数值会增加副本的I/O并发度。 num.io.threads 8 Broker用来处理磁盘I/O的线程数目，这个线程数目建议至少等于硬盘的个数。 KAFKA_HEAP_OPTS -Xmx6G -Xms6G Kafka JVM堆内存设置。当Broker上数据量较大时，应适当调整堆内存大小。

日志级别 Kafka提供了如表4所示的日志级别。 运行日志的级别优先级从高到低分别是ERROR、WARN、INFO、DEBUG，程序会打印高于或等于所设置级别的日志，设置的日志等级越高，打印出来的日志就越少。 表4 日志级别 级别 描述 ERROR ERROR表示系统运行的错误信息。 WARN WARN表示当前事件处理存在异常信息。 INFO INFO表示记录系统及各事件正常运行状态信息。 DEBUG DEBUG表示记录系统及系统的调试信息。 如果您需要修改日志级别，请执行如下操作： 请参考修改集群服务配置参数，进入Kafka的“全部配置”页面。 左边菜单栏中选择所需修改的角色所对应的日志菜单。 选择所需修改的日志级别。 保存配置，在弹出窗口中单击“确定”使配置生效。

操作场景 本章节内容适用于MRS 3.x及后续版本。 增加Broker的存储目录时，MRS集群管理员需要在 FusionInsight Manager中修改Broker的存储目录，以保证Kafka正常工作，新创建的主题分区将在分区最少的目录中生成。适用于以下场景： 由于Kafka不感知磁盘容量，建议各Broker实例配置的磁盘个数和容量保持一致。 更改Broker角色的存储目录，所有Broker实例的存储目录将同步修改。 更改Broker单个实例的存储目录，只对单个实例生效，其他节点Broker实例存储目录不变。

Topic和Partition的划分关系说明 假设集群中部署了K个Kafka节点，每个节点上配置的磁盘个数为N，每块磁盘大小为M，集群共有n个Topic（T1,T2…Tn），并且其中第m个Topic的每秒输入数据总流量为X(Tm) MB/s，配置的副本数为R(Tm)，配置数据保存时间为Y(Tm)小时，那么整体必须满足： 假设单个磁盘大小为M，该磁盘上有n个Partition（P0,P1……Pn），并且其中第m个Partition的每秒写入数据流量为Q(Pm) MB/s（计算方法：所属Topic的数据流量除以Partition数）、数据保存时间为T(Pm)小时，那么单个磁盘必须满足： 根据吞吐量粗略计算，假设生产者可以达到的吞吐量为P，消费者可以达到的吞吐量为C，预期Kafka吞吐量为T，那么建议该Topic的Partition数目设置为Max(T/P , T/C)。 在Kafka集群中，分区越多吞吐量越高，但是分区过多也存在潜在影响，例如文件句柄增加、不可用性增加（如：某个节点故障后，部分Partition重选Leader后时间窗口会比较大）及端到端时延增加等。 建议：单个Partition的磁盘占用最大不超过100GB；单节点上Partition数目不超过3000；整个集群的分区总数不超过10000。

Kafka API简单说明 Producer API 指org.apache.kafka.clients.producer.KafkaProducer中定义的接口，在使用“kafka-console-producer.sh”时，默认使用此API。 Consumer API 指org.apache.kafka.clients.consumer.KafkaConsumer中定义的接口，在使用“kafka-console-consumer.sh”时，默认会调用此API。 MRS 3.x后，Kafka不支持旧Producer API和旧Consumer API。

Kafka访问协议说明 Kafka当前支持四种协议类型的访问：PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。 Kafka服务启动时，默认会启动PLAINTEXT和SASL_PLAINTEXT两种协议类型的访问监测。可通过设置Kafka服务配置“ssl.mode.enable”为“true”，来启动SSL和SASL_SSL两种协议类型的访问监测。下表是四种协议类型的简单说明： 可以参考修改集群服务配置参数进入Kafka全部配置页面，查看或配置参数。 协议类型 说明 默认端口 PLAINTEXT 支持无认证的明文访问。 获取参数“port”的值，默认为9092。 SASL_PLAINTEXT 支持Kerberos认证的明文访问。 获取参数“sasl.port”的值，默认为21007。 SSL 支持无认证的SSL加密访问。 获取参数“ssl.port”的值，默认为9093。 SASL_SSL 支持Kerberos认证的SSL加密访问。 获取参数“sasl-ssl.port”的值，默认为21009。

针对不同的Topic访问场景，Kafka中API使用说明 场景一：访问设置了ACL的Topic 使用的API 用户属组 客户端参数 服务端参数 访问的端口 API 用户需满足以下条件之一即可： 加入System_administrator角色 属于kafkaadmin组 属于kafkasuperuser组 被授权的kafka组的用户 security.inter.broker.protocol=SASL_PLAINTEXT sasl.kerberos.service.name = kafka - sasl.port（默认21007） security.protocol=SASL_SSL sasl.kerberos.service.name = kafka “ssl.mode.enable”配置为true sasl-ssl.port（默认21009） 场景二：访问未设置ACL的Topic 使用的API 用户属组 客户端参数 服务端参数 访问的端口 API 用户需满足以下条件之一： 加入System_administrator角色 属于kafkaadmin组 属于kafkasuperuser组 security.protocol=SASL_PLAINTEXT sasl.kerberos.service.name = kafka - sasl.port（默认21007） 用户属于kafka组 “allow.everyone.if.no.acl.found”配置为true 说明： 普通集群下不涉及服务端参数“allow.everyone.if.no.acl.found”的修改 sasl.port（默认21007） 用户需满足以下条件之一： 加入System_administrator角色 属于kafkaadmin组 kafkasuperuser组用户 security.protocol=SASL_SSL sasl.kerberos.service.name = kafka “ssl.mode.enable”配置为“true” sasl-ssl.port（默认21009） 用户属于kafka组 “allow.everyone.if.no.acl.found”配置为“true” “ssl.mode.enable”配置为“true” sasl-ssl.port（默认21009） - security.protocol=PLAINTEXT “allow.everyone.if.no.acl.found”配置为“true” port（默认9092） - security.protocol=SSL “allow.everyone.if.no.acl.found”配置为“true” “ssl.mode.enable”配置为“true” ssl.port（默认9063）

常用参数 表1 参数说明 配置参数 说明 缺省值 log.dirs Kafka数据存储目录列表，以逗号分隔多个目录。 %{@auto.detect.datapart.bk.log.logs} KAFKA_HEAP_OPTS Kafka启动Broker时使用的jvm选项。建议根据业务需要进行设置。 -Xmx6G -Xms6G auto.create.topics.enable 是否自动创建Topic，如果参数设置为false，发消息前需要通过命令创建Topic。 true default.replication.factor 自动创建Topic时的默认副本数。 2 monitor.preInitDelay 服务启动后，第一次健康检查的延迟时间。如果启动需要较长时间，可以通过调大参数，来完成启动。单位为毫秒。 600000

超时参数 表2 Broker相关超时参数 参数名称 参数说明 默认值 影响分析 controller.socket.timeout.ms Controller连接Broker的超时时间。单位：毫秒。 30000 Controller连接Broker的超时时间，一般不需要调整。 group.max.session.timeout.ms Consumer注册时允许的最大会话超时时间。单位：毫秒。 1800000 允许Consumer配置的session.timeout.ms的最大值（不包含此值）。 group.min.session.timeout.ms Consumer注册时允许的最小会话超时时间。单位：毫秒。 6000 允许Consumer配置的session.timeout.ms的最小值（不包含此值）。 offsets.commit.timeout.ms Offset提交请求的超时时间。单位：毫秒。 5000 Offset提交时被延迟处理的最大超时时间。 replica.socket.timeout.ms 副本数据同步请求的超时时间，配置值不得小于replica.fetch.wait.max.ms。单位：毫秒。 30000 同步线程在发送同步请求之前等待通道建立的最大超时时间，要求配置大于replica.fetch.wait.max.ms。 request.timeout.ms 设置客户端发送连接请求后，等待响应的超时时间。单位：毫秒。 30000 Broker节点上的Controller、Replica线程中传入networkclient连接的超时参数，如果在超时时间内没有接收到响应，那么客户端重新发送，并在达到重试次数后返回请求失败。 transaction.max.timeout.ms 事务允许的最大超时。单位：毫秒。 900000 事务最大超时时间，如果客户端的请求时间超过该值，则Broker将在InitProducerIdRequest中返回一个错误。这样可以防止客户端超时时间过长，而导致消费者无法接收topic。 user.group.cache.timeout.sec 指定缓存中保存用户对应组信息的时间。单位：秒。 300 缓存中用户和组对应关系缓存时间，超过此时间用户信息才会再次通过id -Gn命令查询，在此期间，仅使用缓存中的用户和组对应关系。 zookeeper.connection.timeout.ms 连接ZooKeeper的超时时间。单位：毫秒。 45000 ZooKeeper连接超时时间，这个时间决定了zkclient中初次连接建立过程时允许消耗的时间，超过该时间，zkclient会主动断开。 zookeeper.session.timeout.ms ZooKeeper会话超时时间。如果Broker在此时间内未向ZooKeeper上报心跳，则被认为失效。单位：毫秒。 45000 ZooKeeper会话超时时间。 作用一：这个时间结合传入的ZKURL中ZooKeeper的地址个数，ZooKeeper客户端以（sessionTimeout/传入ZooKeeper地址个数）为连接一个节点的超时时间，超过此时间未连接成功，则尝试连接下一个节点。 作用二：连接建立后，一个会话的超时时间，如ZooKeeper上注册的临时节点BrokerId，当Broker被停止，则该BrokerId，会经过一个sessionTimeout才会被ZooKeeper清理。 表3 Producer相关超时参数 配置名称 说明 默认值 影响分析 request.timeout.ms 指定发送消息请求的请求超时时间。单位：毫秒。 30000 请求超时时间，出现网络问题时，需调大此参数；配置过小，则容易出现Batch Expire异常。 表4 Consumer相关超时参数 配置名称 说明 默认值 影响分析 connections.max.idle.ms 空闲连接的保留时间。单位：毫秒 600000 空闲连接的保留时间，连接空闲时间大于此时间，则会销毁该连接，有需要时重新创建连接。 request.timeout.ms 消费请求的超时时间。单位：毫秒。 30000 请求超时时间，请求超时会失败然后不断重试。

操作步骤 参数入口： HBase角色相关的JVM参数需要配置在安装有HBase服务的节点的“${BIGDATA_HOME}/FusionInsight_HD_*/install/FusionInsight-HBase-2.2.3/hbase/conf/”目录下的“hbase-env.sh”文件中。 每个角色都有各自的JVM参数配置变量，如表1。 表1 HBase相关JVM参数配置变量 变量名 变量影响的角色 HBASE_OPTS 该变量中设置的参数，将影响HBase的所有角色。 SERVER_GC_OPTS 该变量中设置的参数，将影响HBase Server端的所有角色，例如：Master、RegionServer等。 CLIENT_GC_OPTS 该变量中设置的参数，将影响HBase的Client进程。 HBASE_MASTER_OPTS 该变量中设置的参数，将影响HBase的Master。 HBASE_REGIONSERVER_OPTS 该变量中设置的参数，将影响HBase的RegionServer。 HBASE_THRIFT_OPTS 该变量中设置的参数，将影响HBase的Thrift。 配置方式举例： export HADOOP_NAMENODE_OPTS="-Dhadoop.security.logger=${HADOOP_SECURITY_ LOG GER:-INFO,RFAS} -Dhdfs.audit.logger=${HDFS_AUDIT_LOGGER:-INFO,NullAppender} $HADOOP_NAMENODE_OPTS"

操作场景 在Hive中执行多表Join时，Hive支持开启CBO（Cost Based Optimization），系统会自动根据表的统计信息，例如数据量、文件数等，选出合适计划提高多表Join的效率。Hive需要先收集表的统计信息后才能使CBO正确的优化。 CBO优化器会基于统计信息和查询条件，尽可能地使Join顺序达到更优。但是也可能存在特殊情况导致Join顺序调整不准确。例如数据存在倾斜，以及查询条件值在表中不存在等场景，可能调整出非优化的Join顺序。 开启列统计信息自动收集时，需要在Reduce侧做聚合统计。对于没有Reduce阶段的insert任务，将会多出Reduce阶段，用于收集统计信息。

操作步骤 推荐：使用“SNAPPY”压缩，适用于压缩比和读取效率要求均衡场景。 Create table xx (col_name data_type) stored as orc tblproperties ("orc.compress"="SNAPPY"); 可用：使用“ZLIB”压缩，适用于压缩比要求较高场景。 Create table xx (col_name data_type) stored as orc tblproperties ("orc.compress"="ZLIB"); xx为具体使用的Hive表名。

注意事项 Group By数据倾斜 Group By也同样存在数据倾斜的问题，设置“hive.groupby.skewindata”为“true”，生成的查询计划会有两个MapReduce Job，第一个Job的Map输出结果会随机的分布到Reduce中，每个Reduce做聚合操作，并输出结果，这样的处理会使相同的Group By Key可能被分发到不同的Reduce中，从而达到负载均衡，第二个Job再根据预处理的结果按照Group By Key分发到Reduce中完成最终的聚合操作。 Count Distinct聚合问题 当使用聚合函数count distinct完成去重计数时，处理值为空的情况会使Reduce产生很严重的数据倾斜，可以将空值单独处理，如果是计算count distinct，可以通过where子句将该值排除掉，并在最后的count distinct结果中加1。如果还有其他计算，可以先将值为空的记录单独处理，再和其他计算结果合并。