华为云用户手册

操作步骤 在外部节点上确保连接SSH时使用的用户存在，且该用户“~/.ssh”目录存在。 使用omm用户登录Oozie所在节点，查看“~/.ssh/id_rsa.pub”文件是否存在。 是，执行3。 否，执行以下命令生成公私钥： ssh-keygen -t rsa 以omm用户登录oozie实例所在节点，执行以下命令配置互信： ssh-copy-id -i ~/.ssh/id_rsa.pub 运行SSH任务的用户@运行SSH任务的节点的IP地址 执行该命令后需要输入运行SSH任务的用户的密码。 Shell所在节点（外部节点）的账户需要有权限执行Shell脚本并对于所有Shell脚本里涉及到的所有目录文件有足够权限。 如果Oozie具有多个节点，需要在所有Oozie节点执行2~3。 使用omm用户登录依次其他Oozie所在节点，重复执行2-3。

添加安全区 使用Ranger管理员用户rangeradmin登录Ranger管理页面，具体操作可参考登录Ranger WebUI界面。 单击“Security Zone”，在区域列表页面中单击，添加安全区。 表1 安全区配置参数 参数名称 描述 示例 Zone Name 配置安全区的名称。 test Zone Description 配置安全区的描述信息。 - Admin Users/Admin Usergroups 配置安全区的管理用户/用户组，可在安全区中添加及修改相关资源的权限策略。 必须至少配置一个用户或用户组。 zone_admin Auditor Users/ Auditor Usergroups 添加审计用户/用户组，可在安全区中查看相关资源权限策略内容。 必须至少配置一个用户或用户组。 zone_user Select Tag Services 选择服务的标签信息。 - Select Resource Services 选择安全区内包含的服务及具体资源。 在“Select Resource Services”中选择服务后，需要在“Resource”列中添加具体的资源对象，例如HDFS服务器的文件目录、Yarn的队列、Hive的数据库及表、HBase的表及列。 /testzone 例如针对HDFS中的“/testzone”目录创建一个安全区，配置如下： 单击“Save”，等待安全区添加成功。 Ranger管理员可在“Security Zone”页面查看当前的所有安全区并单击“Edit”修改安全区的属性信息，当相关资源不需要在安全区中进行管理时，可单击“Delete”删除对应安全区。

操作步骤 创建工作流，请参考使用Hue创建工作流。 在工作流编辑页面，选择“子Workflow”按钮，将其拖到操作区中。 在弹出的“Sub workflow”窗口中配置“Sub-workflow”的值，例如从下拉列表中选取“Java-Workflow”（这个值是已经创建好的工作流之一），然后单击“添加”。 单击Oozie编辑器右上角的。 保存前如果需要修改作业名称（默认为“My Workflow”），可以直接单击该名称进行修改，例如“Subworkflow-Workflow”。 保存完成后，单击，提交该作业。 作业提交后，可通过Hue界面查看作业的详细信息、日志、进度等相关内容。

操作步骤 创建工作流，请参考使用Hue创建工作流。 在工作流编辑页面，选择“Spark 程序”按钮，将其拖到操作区中。 在弹出的“Spark”窗口配置“Files”，例如“hdfs://hacluster/user/admin/examples/apps/spark2x/lib/oozie-examples.jar”。配置“jar/py name”，例如“oozie-examples.jar” ，配置完成后单击“添加”。 配置“Main class”的值。例如“org.apache.oozie.example.SparkFileCopy”。 单击“参数+”，添加输入输出相关参数。 例如添加： “hdfs://hacluster/user/admin/examples/input-data/text/data.txt” “hdfs://hacluster/user/admin/examples/output-data/spark_workflow” 在“Options list”文本框指定spark参数 , 例如“--conf spark.yarn.archive=hdfs://hacluster/user/spark2x/jars/8.1.0.1/spark-archive-2x.zip --conf spark.eventLog.enabled=true --conf spark.eventLog.dir=hdfs://hacluster/spark2xJobHistory2x”。 此处版本号“8.1.0.1”为示例，可登录 FusionInsight Manager界面，单击右上角的，在下拉框中单击“关于”，在弹框中查看Manager版本号。 单击右上角的配置按钮。配置“Spark Master”的值，例如“yarn-cluster”。配置“Mode”的值，例如“cluster”。 在打开的配置界面中，单击“删除+”，添加删除目录，例如“hdfs://hacluster/user/admin/examples/output-data/spark_workflow”。 单击“属性+”，添加oozie使用的sharelib，左边文本框填写属性名称“oozie.action.sharelib.for.spark”，右边文本框填写属性值“spark2x”。 单击Oozie编辑器右上角的。 保存前如果需要修改作业名称（默认为“My Workflow”），可以直接单击该名称进行修改，例如“Spark-Workflow”。 保存完成后，单击，提交该作业。 作业提交后，可通过Hue界面查看作业的详细信息、日志、进度等相关内容。

配置场景 Spark中见到的UI、EventLog、动态资源调度等功能都是通过事件传递实现的。事件有SparkListenerJobStart、SparkListenerJobEnd等，记录了每个重要的过程。 每个事件在发生后都会保存到一个队列中，Driver在创建SparkContext对象时，会启动一个线程循环的从该队列中依次拿出一个事件，然后发送给各个Listener，每个Listener感知到事件后就会做各自的处理。 因此当队列存放的速度大于获取的速度时，就会导致队列溢出，从而丢失了溢出的事件，影响了UI、EventLog、动态资源调度等功能。所以为了更灵活的使用，在这边添加一个配置项，用户可以根据Driver的内存大小设置合适的值。

配置描述 提供两种不同的数据汇聚功能配置选项，两者在Spark JD BCS erver服务端的tunning选项中进行设置，设置完后需要重启JDB CS erver。 表1 参数说明 参数 说明 默认值 spark.sql.bigdata.thriftServer.useHdfsCollect 是否将结果数据保存到HDFS中而不是内存中。 优点：由于查询结果保存在hdfs端，因此基本不会造成JDBCServer的OOM。 缺点：速度慢。 true：保存至HDFS中 false：不使用该功能 须知： spark.sql.bigdata.thriftServer.useHdfsCollect参数设置为true时，将结果数据保存到HDFS中，但JobHistory原生页面上Job的描述信息无法正常关联到对应的SQL语句，同时spark-beeline命令行中回显的Execution ID为null，为解决JDBCServer OOM问题，同时显示信息正确，建议选择 spark.sql.userlocalFileCollect参数进行配置。 false spark.sql.uselocalFileCollect 是否将结果数据保存在本地磁盘中而不是内存里面。 优点：结果数据小数据量情况下和原生内存的方式相比性能损失可以忽略，大数据情况下（亿级数据）性能远比使用hdfs，以及原生内存方式好。 缺点：需要调优。大数据情况下建议JDBCServer driver端内存10G，executor端每个核心分配3G内存。 true：使用该功能 false: 不使用该功能 false spark.sql.collect.Hive 该参数在spark.sql.uselocalFileCollect开启的情况下生效。直接序列化的方式，还是间接序列化的方式保存结果数据到磁盘。 优点：针对分区数特别多的表查询结果汇聚性能优于直接使用结果数据保证在磁盘的方式。 缺点：和spark.sql.uselocalFileCollect开启时候的缺点一样。 true：使用该功能 false：不使用该功能 false spark.sql.collect.serialize 该参数在spark.sql.uselocalFileCollect， spark.sql.collect.Hive同时开启的情况下生效。 作用是进一步提升性能 java：采用java序列化方式收集数据。 kryo：采用kryo序列化方式收集数据，性能要比采用java好。 java 参数spark.sql.bigdata.thriftServer.useHdfsCollect和spark.sql.uselocalFileCollect不能同时设置为true。

配置场景 当Spark2x Web UI中有一些不允许其他用户看到的数据时，用户可能想对UI进行安全防护。用户一旦登录，Spark2x可以比较与这个用户相对应的视图ACLs来确认是否授权用户访问 UI。 Spark2x存在两种类型的Web UI，一种为运行中任务的Web UI，可以通过Yarn原生页面的应用链接或者REST接口访问。一种为已结束任务的Web UI，可以通过Spark2x JobHistory服务或者REST接口访问。 本章节仅支持安全模式（开启了Kerberos认证）集群。 运行中任务Web UI ACL配置。 运行中的任务，可通过服务端对如下参数进行配置。 “spark.admin.acls”：指定Web UI的管理员列表。 “spark.admin.acls.groups”：指定管理员组列表。 “spark.ui.view.acls”：指定yarn界面的访问者列表。 “spark.modify.acls.groups”：指定yarn界面的访问者组列表。 “spark.modify.acls”：指定Web UI的修改者列表。 “spark.ui.view.acls.groups”：指定Web UI的修改者组列表。 运行结束后Web UI ACL配置。 运行结束的任务通过客户端的参数“spark.history.ui.acls.enable”控制是否开启ACL访问权限。 如果开启了ACL控制，由客户端的“spark.admin.acls”和“spark.admin.acls.groups”配置指定Web UI的管理员列表和管理员组列表，由客户端的“spark.ui.view.acls”和“spark.modify.acls.groups”配置指定查看Web UI任务明细的访问者列表和组列表，由客户端的“spark.modify.acls”和“spark.ui.view.acls.groups”配置指定修改Web UI任务明细的访问者列表和组列表。

场景说明 假定用户有某个周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Spark应用程序实现如下要求： 统计日志文件中本周末网购停留总时间超过2个小时的女性网民信息。 周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为“,”。 log1.txt：周六网民停留日志 LiuYang,female,20YuanJing,male,10GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20FangBo,female,50LiuYang,female,20YuanJing,male,10GuoYijun,male,50CaiXuyu,female,50FangBo,female,60 log2.txt：周日网民停留日志 LiuYang,female,20YuanJing,male,10CaiXuyu,female,50FangBo,female,50GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20CaiXuyu,female,50FangBo,female,50LiuYang,female,20YuanJing,male,10FangBo,female,50GuoYijun,male,50CaiXuyu,female,50FangBo,female,60

查看Ranger审计信息内容 使用Ranger管理员用户rangeradmin登录Ranger管理页面，具体操作可参考登录Ranger WebUI界面。 单击“Audit”，查看相关审计信息，各页签内容说明请参考表1，条目较多时，单击搜索框可根据关键字字段进行筛选。 表1 Audit信息 页签 内容描述 Access 当前 MRS 不支持在线查看组件资源的审计日志信息，可登录组件安装节点，进入“/var/log/Bigdata/audit”目录下查看各组件的审计日志。 Admin Ranger上操作审计信息，例如安全访问策略的创建/更新/删除、组件权限策略的创建/删除、role的创建/更新/删除等。 Login Sessions 登录Ranger的用户会话审计信息。 Plugins Ranger内组件权限策略信息。 Plugin Status 各组件节点权限策略的同步审计信息。 User Sync Ranger与LDAP用户同步审计信息。

操作步骤 访问Hue WebUI，请参考访问Hue WebUI界面。 在界面左侧导航栏单击，选择“Bundle”，打开Bundle编辑器。 在作业编辑界面中单击“My Bundle”修改作业的名称。 单击“+添加Coordinator”选择需要编排的Coordinator作业。 根据界面提示设置Coordinator任务调度的开始、结束时间，然后单击右上角的保存作业。 单击编辑器右上角的，在弹出菜单选择，设置Bundle任务的启动时间，根据实际需求单击“+添加参数”设置提交参数，然后关闭对话框保存设置。 因时区转化的原因，此处时间有可能会与当地系统实际时间差异数个小时。比如在中国，此处的时间则会比当地时间晚8个小时。 单击编辑器右上角的，在弹出的确认界面中单击“提交”提交作业。

Ranger用户类型 Ranger中的用户可分为Admin、User、Auditor等类型，不同用户具有的Ranger管理界面查看和操作权限不同。 Admin：Ranger安全管理员，可查看Ranger所有管理页面内容，进行服务权限管理插件及权限访问控制策略的管理操作，可查看审计信息内容，可进行用户类型设置。 Auditor：Ranger审计管理员，可查看服务权限管理插件及权限访问控制策略的内容。 User：普通用户，可以被Ranger管理员赋予具体权限。

操作步骤 以客户端安装用户，登录安装Oozie客户端的节点。 执行以下命令，获取安装环境信息。其中“/opt/client”为客户端安装路径，该操作的客户端目录只是举例，请根据实际安装目录修改。 source /opt/client/bigdata_env 判断集群认证模式。 安全模式，执行kinit命令进行用户认证。 例如，使用oozieuser用户进行认证。 kinit oozieuser 普通模式，执行4。 执行以下命令，进入样例目录。 cd /opt/client/Oozie/oozie-client-*/examples/apps/sqoop/ 该目录下需关注文件如表1所示。 表1 文件说明 文件名称 描述 job.properties 工作流的参数变量定义文件。 workflow.xml 工作流的规则定制文件。 执行以下命令，编辑“job.properties”文件。 vi job.properties 修改如下内容： 更改“userName”的参数值为提交任务的人机用户名，例如“userName=oozieuser”。 执行以下命令，编辑“workflow.xml”文件。 vi workflow.xml 修改如下内容： “command”的值修改为需要调度的已有Loader作业ID，例如1。 将“workflow.xml”文件上传至 "job.properties" 文件中的HDFS路径。 hdfs dfs -put -f workflow.xml /user/userName/examples/apps/sqoop 执行oozie job命令，运行工作流文件。 oozie job -oozie https://oozie角色的主机名:21003/oozie/ -config job.properties -run 命令参数解释如下： -oozie：实际执行任务的Oozie服务器URL。 -config：工作流属性文件。 -run：运行工作流。 执行完工作流文件，显示job id表示提交成功，例如：job: 0000021-140222101051722-oozie-omm-W。登录Oozie管理页面，查看运行情况。 使用oozieuser用户，登录Oozie WebUI页面：https://oozie角色的ip地址:21003/oozie 。 Oozie的WebUI界面中，可在页面表格根据jobid查看已提交的工作流信息。

前提条件 Loader和Oozie组件及客户端已经安装，并且正常运行。 已创建或获取访问Oozie服务的人机用户账号及密码。 该用户需要从属于hadoop、supergroup、hive组，同时添加Oozie的角色操作权限。若使用Hive多实例，该用户还需要从属于具体的Hive实例组，如hive3。 用户同时还需要至少有manager_viewer权限的角色。 获取运行状态的Oozie服务器（任意实例）URL，如“https://10.1.130.10:21003/oozie”。 获取运行状态的Oozie服务器主机名，如“10-1-130-10”。 获取Yarn ResourceManager主节点IP，如10.1.130.11。 创建需要调度的Loader作业，并获取该作业ID。

操作步骤 创建工作流，请参考使用Hue创建工作流。 在工作流编辑页面，选择“数据流”按钮，将其拖到操作区中。 在弹出的“Streaming”窗口中配置“Mapper”的值，例如“/bin/cat”。配置“Reducer”的值，例如“/usr/bin/wc”。然后单击“添加”。 单击“文件+”，添加运行所需的文件。 例如“/user/oozie/share/lib/mapreduce-streaming/hadoop-streaming-xxx.jar”和“/user/oozie/share/lib/mapreduce-streaming/oozie-sharelib-streaming-5.1.0.jar”。 单击右上角的配置按钮。在打开的配置界面中，单击“删除+”，添加删除目录，例如“/user/admin/examples/output-data/streaming_workflow”。 单击“属性+”，添加下列属性。 左边框填写属性名称“mapred.input.dir”，右边框填写属性值“/user/admin/examples/input-data/text”。 左边框填写属性名称“mapred.output.dir”，右边框填写属性值“/user/admin/examples/output-data/streaming_workflow”。 单击Oozie编辑器右上角的。 保存前如果需要修改作业名称（默认为“My Workflow”），可以直接单击该名称进行修改，例如“Streaming-Workflow”。 保存完成后，单击，提交该作业。 作业提交后，可通过Hue界面查看作业的详细信息、日志、进度等相关内容。

日志描述 日志存储路径： Executor运行日志：“${BIGDATA_DATA_HOME}/hadoop/data${i}/nm/containerlogs/application_${appid}/container_{$contid}” 运行中的任务日志存储在以上路径中，运行结束后会基于Yarn的配置确定是否汇聚到HDFS目录中，详情请参见Yarn常用配置参数。 其他日志：“/var/log/Bigdata/spark2x” 日志归档规则： 使用yarn-client或yarn-cluster模式提交任务时，Executor日志默认50MB滚动存储一次，最多保留10个文件，不压缩。 JobHistory2x日志默认100MB滚动存储一次，最多保留100个文件，压缩存储。 JDBCServer2x日志默认100MB滚动存储一次，最多保留100个文件，压缩存储。 IndexServer2x日志默认100MB滚动存储一次，最多保留100个文件，压缩存储。 JDBCServer2x审计日志默认20MB滚动存储一次，最多保留20个文件，压缩存储。 日志大小和压缩文件保留个数可以在FusionInsight Manager界面中配置。 表1 Spark2x日志列表 日志类型 日志文件名 描述 SparkResource2x日志 spark.log Spark2x服务初始化日志。 prestart.log prestart脚本日志。 cleanup.log 安装卸载实例时的清理日志。 spark-availability-check.log Spark2x服务健康检查日志。 spark-service-check.log Spark2x服务检查日志 JDBCServer2x日志 JDBCServer-start.log JDBCServer2x启动日志。 JDBCServer-stop.log JDBCServer2x停止日志。 JDBCServer.log JDBCServer2x运行时，Driver端日志。 jdbc-state-check.log JDBCServer2x健康检查日志。 jdbcserver-omm-pid***-gc.log.*.current JDBCServer2x进程gc日志。 spark-omm-org.apache.spark.sql.hive.thriftserver.HiveThriftProxyServer2-***.out* JDBCServer2x进程启动信息日志。若进程停止，会打印jstack信息。 JobHistory2x日志 jobHistory-start.log JobHistory2x启动日志。 jobHistory-stop.log JobHistory2x停止日志。 JobHistory.log JobHistory2x运行过程日志。 jobhistory-omm-pid***-gc.log.*.current JobHistory2x进程gc日志。 spark-omm-org.apache.spark.deploy.history.HistoryServer-***.out* JobHistory2x进程启动信息日志。若进程停止，会打印jstack信息。 IndexServer2x日志 IndexServer-start.log IndexServer2x启动日志。 IndexServer-stop.log IndexServer2x停止日志。 IndexServer.log IndexServer2x运行时，Driver端日志。 indexserver-state-check.log IndexServer2x健康检查日志。 indexserver-omm-pid***-gc.log.*.current IndexServer2x进程gc日志。 spark-omm-org.apache.spark.sql.hive.thriftserver.IndexServerProxy-***.out* IndexServer2x进程启动信息日志。若进程停止，会打印jstack信息。 审计日志 jdbcserver-audit.log ranger-audit.log JDBCServer2x审计日志。

操作步骤 创建工作流，请参考使用Hue创建工作流。 在工作流编辑页面，选择“HiveServer2 脚本”按钮，将其拖到操作区中。 在弹出的“HiveServer2 Script”窗口中配置HDFS上的脚本路径，例如“/user/admin/examples/apps/hive2/script.q”，然后单击“添加”。 单击“参数+”，添加输入输出参数。 例如输入参数为“INPUT=/user/admin/examples/input-data/table”，输出参数为“OUTPUT=/user/admin/examples/output-data/hive2_workflow”。 单击右上角的配置按钮。在打开的配置界面中，单击“删除+”，添加删除目录，例如“/user/admin/examples/output-data/hive2_workflow”。 配置“作业 XML”，值为“客户端安装目录/Oozie/oozie-client-*/examples/apps/hive/hive-site.xml”上传至HDFS目录中所在路径，例如“/user/admin/examples/apps/hive2/hive-site.xml”。“HiveServer2 URL”及其他参数无需配置。 若以上的参数和值在使用过程中发生了修改，可在“Oozie客户端安装目录/oozie-client-*/conf/hive-site.xml”文件中查询。 单击Oozie编辑器右上角的。 保存前如果需要修改作业名称（默认为“My Workflow”），可以直接单击该名称进行修改，例如“Hive2-Workflow”。 保存完成后，单击，提交该作业。 作业提交后，可通过Hue界面查看作业的详细信息、日志、进度等相关内容。

操作步骤 创建工作流，请参考使用Hue创建工作流。 在工作流编辑页面，选择“Shell”按钮，将其拖到操作区中。 在弹出的“Shell”窗口中配置“Shell command”的值，例如“oozie_shell.sh”，然后单击“添加”。 单击“文件+”，添加Shell命令执行文件或Oozie样例执行文件，可以选择存储在HDFS的文件或本地文件。 若文件存储在HDFS上，选择“.sh”文件所在路径即可，例如“user/hueuser/shell/oozie_shell.sh”。 若选择本地文件，则需在“选择文件”界面，单击“上传文件”，上传本地文件，文件上传成功后，选择该文件即可。 如果执行的Shell文件需要传递参数，可单击“参数+”设置参数。 传递参数的顺序需要和Shell脚本中保持一致。 单击Oozie编辑器右上角的。 保存前如果需要修改作业名称（默认为“My Workflow”），可以直接单击该名称进行修改，例如“Shell-Workflow”。 保存完成后，单击，提交该作业。 作业提交后，可通过Hue界面查看作业的详细信息、日志、进度等相关内容。 配置Shell命令为Linux指令时，请指定为原始指令，不要使用快捷键指令。例如：ls -l ,不要配置成ll。可配置成Shell命令ls ，参数添加一个 “-l”。 Windows上传Shell脚本到HDFS时，请保证Shell脚本的格式为Unix，格式不正确会导致Shell作业提交失败。

操作步骤 创建工作流，请参考使用Hue创建工作流。 在工作流编辑页面，选择“Java 程序”按钮，将其拖到操作区中。 在弹出的“Java program”窗口中配置“Jar name”的值，例如“/user/admin/examples/apps/java-main/lib/oozie-examples-5.1.0.jar”。配置“Main class”的值，例如“org.apache.oozie.example.DemoJavaMain”。然后单击“添加”。 单击Oozie编辑器右上角的。 保存前如果需要修改作业名称（默认为“My Workflow”），可以直接单击该名称进行修改，例如“Java-Workflow”。 保存完成后，单击，提交该作业。 作业提交后，可通过Hue界面查看作业的详细信息、日志、进度等相关内容。

配置场景 当Yarn配置“yarn.log-aggregation-enable”为“true”时，就开启了container日志聚合功能。日志聚合功能是指：当应用在Yarn上执行完成后，NodeManager将本节点中所有container的日志聚合到HDFS中，并删除本地日志。详情请参见配置Container日志聚合功能。 然而，开启container日志聚合功能之后，其日志聚合至HDFS目录中，只能通过获取HDFS文件来查看日志。开源Spark和Yarn服务不支持通过WebUI查看聚合后的日志。 因此，Spark在此基础上进行了功能增强。如图1所示，在HistoryServer页面添加“AggregatedLogs”页签，可以通过“logs”链接查看聚合的日志。 图1 聚合日志显示页面

配置描述 为了使WebUI页面显示日志，需要将聚合日志进行解析和展现。Spark是通过Hadoop的JobHistoryServer来解析聚合日志的，所以您可以通过“spark.jobhistory.address”参数，指定JobHistoryServer页面地址，即可完成解析和展现。 参数入口： 在应用提交时通过“--conf”设置这些参数，或者在客户端的“spark-defaults.conf”配置文件中调整如下参数。 此功能依赖Hadoop中的JobHistoryServer服务，所以使用聚合日志之前需要保证JobHistoryServer服务已经运行正常。 如果参数值为空，“AggregatedLogs”页签仍然存在，但是无法通过logs链接查看日志。 只有当App已经running，HDFS上已经有该App的事件日志文件时才能查看到聚合的container日志。 正在运行的任务的日志，用户可以通过“Executors”页面的日志链接进行查看，任务结束后日志会汇聚到HDFS上，“Executors”页面的日志链接就会失效，此时用户可以通过“AggregatedLogs”页面的logs链接查看聚合日志。 表1 参数说明 参数 描述 默认值 spark.jobhistory.address JobHistoryServer页面的地址，格式：http(s)://ip:port/jobhistory。例如，将参数值设置为“https://10.92.115.1:26014/jobhistory”。 默认值为空，表示不能从WebUI查看container聚合日志。 修改参数后，需重启服务使得配置生效。 -

配置场景 Spark Streaming对接Kafka时，当Spark Streaming应用重启后，应用根据上一次读取的topic offset作为起始位置和当前topic最新的offset作为结束位置从Kafka上读取数据的。 Kafka服务的topic的leader异常后，若Kafka的leader和follower的offset相差太大，用户重启Kafka服务，Kafka的follower和leader相互切换，则Kafka服务重启后，topic的offset变小。 若Spark Streaming应用一直在运行，由于Kafka上topic的offset变小，会导致读取Kafka数据的起始位置比结束位置大，这样将无法从Kafka读取数据，应用报错。 若在重启Kafka服务前，先停止Spark Streaming应用，等Kafka重启后，再重启Spark Streaming应用使应用从checkpoint恢复。此时，Spark Streaming应用会记录终止前读取到的offset位置，以此为基准读取后面的数据，而Kafka offset变小（例如从10万变成1万），Spark Streaming会等待Kafka leader的offset增长至10万之后才会去消费，导致新发送的offset在1万至10万之间的数据丢失。 针对上述背景，提供配置Streaming对接Kafka更高级别的可靠性。对接Kafka可靠性功能开启后，上述场景处理方式如下。 若Spark Streaming应用在运行应用时Kafka上topic的offset变小，则会将Kafka上topic最新的offset作为读取Kafka数据的起始位置，继续读取后续的数据。 对于已经生成但未调度处理的任务，若读取的Kafka offset区间大于Kafka上topic的最新offset，则该任务会运行失败。 若任务失败过多，则会将executor加入黑名单，从而导致后续的任务无法部署运行。此时用户可以通过配置“spark.blacklist.enabled”参数关闭黑名单功能，黑名单功能默认为开启。 若Kafka上topic的offset变小后，Spark Streaming应用进行重启恢复终止前未处理完的任务若读取的Kafka offset区间大于Kafka上topic的最新offset，则该任务直接丢弃，不进行处理。 若Streaming应用中使用了state函数，则不允许开启对接Kafka可靠性功能。

配置场景 当Spark Streaming应用与Kafka对接，Spark Streaming应用异常终止并从checkpoint恢复重启后，对于进入Kafka数据的任务，系统默认优先处理应用终止前（A段时间）未完成的任务和应用终止到重启完成这段时间内（B段时间）进入Kafka数据生成的任务，最后再处理应用重启完成后（C段时间）进入Kafka数据生成的任务。并且对于B段时间进入Kafka的数据，Spark将按照终止时间（batch时间）生成相应个数的任务，其中第一个任务读取全部数据，其余任务可能不读取数据，造成任务处理压力不均匀。 若A段时间的任务和B段时间任务处理得较慢，则会影响C段时间任务的处理。针对上述场景，Spark提供Kafka后进先出功能。 图1 Spark Streaming应用重启时间轴 开启此功能后，Spark将优先调度C段时间内的任务，若存在多个C段任务，则按照任务产生的先后顺序调度执行，再执行A段时间和B段时间的任务。另外，对于B段时间进入Kafka的数据，Spark除了按照终止时间生成相应任务，还将这个期间进入Kafka的所有数据均匀分配到各个任务，避免任务处理压力不均匀。 约束条件： 目前该功能只适用于Spark Streaming中的Direct方式，且执行结果与上一个batch时间处理结果没有依赖关系（即无state操作，如updatestatebykey）。对多条数据输入流，需要相对独立无依赖的状态，否则可能导致数据切分后结果发生变化。 Kafka后进先出功能的开启要求应用只能对接Kafka输入源。 若提交应用的同时开启Kafka后进先出和流控功能，对于B段时间进入Kafka的数据，将不启动流控功能，以确保读取这些数据的任务调度优先级最低。应用重新启动后C段时间的任务启用流控功能。

配置描述 在Spark Driver端的“spark-defaults.conf”配置文件中进行设置。 表1 参数说明 参数 说明 默认值 spark.streaming.kafka.direct.lifo 配置是否开启Kafka后进先出功能。 false spark.streaming.kafka010.inputstream.class 获取解耦在FusionInsight侧的类 org.apache.spark.streaming.kafka010.HWDirectKafkaInputDStream

SparkSQL权限 类似于Hive，SparkSQL也是建立在Hadoop上的 数据仓库 框架，提供类似SQL的结构化数据。 MRS提供用户、用户组和角色，集群中的各类权限需要先授予角色，然后将用户或者用户组与角色绑定。用户只有绑定角色或者加入绑定角色的用户组，才能获得权限。 如果当前组件使用了Ranger进行权限控制，须基于Ranger配置相关策略进行权限管理，具体操作可参考添加Spark2x的Ranger访问权限策略。 Spark2x开启或关闭Ranger鉴权后，需要重启Spark2x服务，并重新下载客户端，或刷新客户端配置文件spark/conf/spark-defaults.conf： 开启Ranger鉴权：spark.ranger.plugin.authorization.enable=true 关闭Ranger鉴权：spark.ranger.plugin.authorization.enable=false

SparkSQL权限模型 用户使用SparkSQL服务进行SQL操作，必须对SparkSQL数据库和表（含外表和视图）拥有相应的权限。完整的SparkSQL权限模型由元数据权限与HDFS文件权限组成。使用数据库或表时所需要的各种权限都是SparkSQL权限模型中的一种。 元数据权限 元数据权限即在元数据层上进行权限控制，与传统关系型数据库类似，SparkSQL数据库包含“创建”和“查询”权限，表和列包含“查询”、“插入”、“UPDATE”和“删除”权限。SparkSQL中还包含拥有者权限“OWNERSHIP”和Spark管理员权限“管理”。 数据文件权限，即HDFS文件权限 SparkSQL的数据库、表对应的文件保存在HDFS中。默认创建的数据库或表保存在HDFS目录“/user/hive/warehouse”。系统自动以数据库名称和数据库中表的名称创建子目录。访问数据库或者表，需要在HDFS中拥有对应文件的权限，包含“读”、“写”和“执行”权限。 用户对SparkSQL数据库或表执行不同操作时，需要关联不同的元数据权限与HDFS文件权限。例如，对SparkSQL数据表执行查询操作，需要关联元数据权限“查询”，以及HDFS文件权限“读”和“执行”。 使用Manager界面图形化的角色管理功能来管理SparkSQL数据库和表的权限，只需要设置元数据权限，系统会自动关联HDFS文件权限，减少界面操作，提高效率。

SparkSQL使用场景及对应权限 用户通过SparkSQL服务创建数据库需要加入Hive组，不需要角色授权。用户在Hive和HDFS中对自己创建的数据库或表拥有完整权限，可直接创建表、查询数据、删除数据、插入数据、更新数据以及授权他人访问表与对应HDFS目录与文件。 如果用户访问别人创建的表或数据库，需要授予权限。所以根据SparkSQL使用场景的不同，用户需要的权限可能也不相同。 表1 SparkSQL使用场景 主要场景 用户需要的权限 使用SparkSQL表、列或数据库 使用其他用户创建的表、列或数据库，不同的场景需要不同的权限，例如： 创建表，需要“创建”。 查询数据，需要“查询”。 插入数据，需要“插入”。 关联使用其他组件 部分场景除了SparkSQL权限，还可能需要组件的权限，例如： 使用Spark on HBase，在SparkSQL中查询HBase表数据，需要设置HBase权限。 在一些特殊SparkSQL使用场景下，需要单独设置其他权限。 表2 SparkSQL授权注意事项 场景 用户需要的权限 创建SparkSQL数据库、表、外表，或者为已经创建的表或外表添加分区，且Hive用户指定数据文件保存在“/user/hive/warehouse”以外的HDFS目录。 需要此目录已经存在，客户端用户是目录的属主，且用户对目录拥有“读”、“写”和“执行”权限。同时用户对此目录上层的每一级目录都拥有“读”和“执行”权限。 在Spark2x中，在创建HBase的外表时，需要拥有Hive端database的“创建”权限。而在Spark 1.5中，在创建HBase的外表时，需要拥有Hive端database的“创建”权限，也需要拥有HBase端Namespace的“创建”权限。 用户使用load将指定目录下所有文件或者指定文件，导入数据到表中。 数据源为Linux本地磁盘，指定目录时需要此目录已经存在，系统用户“omm”对此目录以及此目录上层的每一级目录拥有“r”和“x”的权限。指定文件时需要此文件已经存在，“omm”对此文件拥有“r”的权限，同时对此文件上层的每一级目录拥有“r”和“x”的权限。 数据源为HDFS，指定目录时需要此目录已经存在，SparkSQL用户是目录属主，且用户对此目录及其子目录拥有“读”、“写”和“执行”权限，并且其上层的每一级目录拥有“读”和“执行”权限。指定文件时需要此文件已经存在，SparkSQL用户是文件属主，且用户对文件拥有“读”、“写”和“执行”权限，同时对此文件上层的每一级目录拥有“读”和“执行”权限。 创建函数、删除函数或者修改任意数据库。 需要授予“管理”权限。 操作Hive中所有的数据库和表。 需加入到supergroup用户组，并且授予“管理”权限。 对部分datasource表赋予insert权限后，执行insert|analyze操作前需要单独对hdfs上的表目录赋予写权限。 当前对spark datasource表赋予Insert权限时，若表格式为：text|csv|json|parquet|orc,则不会修改表目录的权限。因此，对以上几种类型的datasource表赋予Insert权限后，还需要单独对hdfs上的表目录赋予写权限，用户才能成功对表执行insert|analyze操作。

操作步骤 使用Ranger管理员用户rangeradmin登录Ranger管理页面，具体操作可参考登录Ranger WebUI界面。 在首页中单击“STORM”区域的“Storm”。 单击“Add New Policy”，添加Storm权限控制策略。 根据业务需求配置相关参数。 表1 Storm权限参数 参数名称 描述 Policy Conditions IP过滤策略，可自定义，配置当前策略适用的主机节点，可填写一个或多个IP或IP段，并且IP填写支持“*”通配符，例如：192.168.1.10,192.168.1.20或者192.168.1.*。 Policy Name 策略名称，可自定义，不能与本服务内其他策略名称重复。 “include”策略适用于当前输入的对象，“exclude”表示策略适用于除去当前输入内容之外的其他对象。 Policy Label 为当前策略指定一个标签，您可以根据这些标签搜索报告和筛选策略。 Storm Topology 配置当前策略适用的拓扑名称。可以填写多个值。 Description 策略描述信息。 Audit Logging 是否审计此策略。 Allow Conditions 策略允许条件，配置本策略内允许的权限及例外。 在“Select Role”、“Select Group”、“Select User”列选择已创建好的需要授予权限的Role、用户组或用户，单击“Add Conditions”，添加策略适用的IP地址范围，单击“Add Permissions”，添加对应权限。 Submit Topology：提交拓扑。 说明： Submit Topology权限只有在Storm Topology为*的情况下可以赋权生效。 File Upload：文件上传。 File Download：文件下载。 Kill Topology：删除拓扑。 Rebalance：Rebalance操作权限。 Activate：激活权限。 Deactivate：去激活权限。 Get Topology Conf：获取拓扑配置。 Get Topology：获取拓扑。 Get User Topology：获取用户拓扑。 Get Topology Info：获取拓扑信息。 Upload New Credential：上传新的凭证。 Select/Deselect All：全选/取消全选。 如需添加多条权限控制规则，可单击按钮添加。 如需当前条件中的用户或用户组管理本条策略，可勾选“Delegate Admin”，这些用户将成为受委托的管理员。被委托的管理员可以更新、删除本策略，它还可以基于原始策略创建子策略。 Deny Conditions 策略拒绝条件，配置本策略内拒绝的权限及例外，配置方法与“Allow Conditions”类似。 （可选）添加策略有效期。在页面右上角单击“Add Validity period”，设置“Start Time”和“End Time”，选择“Time Zone”。单击“Save”保存。如需添加多条策略有效期，可单击按钮添加。如需删除策略有效期，可单击按钮删除。 单击“Add”，在策略列表可查看策略的基本信息。等待策略生效后，验证相关权限是否正常。 如需禁用某条策略，可单击按钮编辑策略，设置策略开关为“Disabled”。 如果不再使用策略，可单击按钮删除策略。

配置场景 当前，在YARN-Client和YARN-Cluster模式下，两种模式的客户端存在冲突的配置，即当客户端为一种模式的配置时，会导致在另一种模式下提交任务失败。 为避免出现如上情况，添加表1中的配置项，避免两种模式下来回切换参数，提升软件易用性。 YARN-Cluster模式下，优先使用新增配置项的值，即服务端路径和参数。 YARN-Client模式下，直接使用原有的三个配置项的值。 原有的三个配置项为：“spark.driver.extraClassPath”、“spark.driver.extraJavaOptions”、“spark.driver.extraLibraryPath”。 不添加表1中配置项时，使用方式与原有方式一致，程序可正常执行，只是在不同模式下需切换配置。

操作步骤 SparkSQL表授权、列授权、数据库授权与Hive的操作相同，详情请参见Hive用户权限管理。 在权限管理中，为了方便用户使用，授予数据库下表的任意权限将自动关联该数据库目录的HDFS权限。为了避免产生性能问题，取消表的任意权限，系统不会自动取消数据库目录的HDFS权限，但对应的用户只能登录数据库和查看表名。 若为角色添加或删除数据库的查询权限，数据库中的表也将自动添加或删除查询权限。此机制为Hive实现，SparkSQL与Hive保持一致。 Spark不支持struct数据类型中列名称含有特殊字符（除字母、数字、下划线外的其他字符）。如果struct类型中列名称含有特殊字符，在FusionInsight Manager的“编辑角色”页面进行授权时，该列将无法正确显示。

相关概念 SparkSQL的语句在SparkSQL中进行处理，权限要求如表1所示。 表1 使用SparkSQL表、列或数据库场景权限一览 操作场景 用户需要的权限 CREATE TABLE “创建”，RWX+ownership（for create external table - the location） 说明： 按照指定文件路径创建datasource表时，需要path后面文件的RWX+ownership权限。 DROP TABLE “Ownership”（of table） DROP TABLE PROPERTIES “Ownership” DESCRIBE TABLE “查询” SHOW PARTITIONS “查询” ALTER TABLE LOCATION “Ownership”，RWX+ownership (for new location) ALTER PARTITION LOCATION “Ownership”，RWX+ownership (for new partition location) ALTER TABLE ADD PARTITION “插入”，RWX+ownership (for partition location) ALTER TABLE DROP PARTITION “删除” ALTER TABLE(all of them except the ones above) “Update”，“Ownership” TRUNCATE TABLE “Ownership” CREATE VIEW “查询”，“Grant Of Select”，“创建” ALTER VIEW PROPERTIES “Ownership” ALTER VIEW RENAME “Ownership” ALTER VIEW ADD PARTS “Ownership” ALTER VIEW AS “Ownership” ALTER VIEW DROPPARTS “Ownership” ANALYZE TABLE “查询”，“插入” SHOW COLUMNS “查询” SHOW TABLE PROPERTIES “查询” CREATE TABLE AS SELECT “查询”，“创建” SELECT “查询” 说明： 与表一样，对视图进行SELECT操作的时候需要有该视图的“查询”权限。 INSERT “插入”，“删除 (for overwrite)” LOAD “插入”，“删除”，RWX+ownership(input location) SHOW CREATE TABLE “查询”，“Grant Of Select” CREATE FUNCTION “管理” DROP FUNCTION “管理” DESC FUNCTION - SHOW FUNCTIONS - MSCK (metastore check) “Ownership” ALTER DATABASE “管理” CREATE DATABASE - SHOW DATABASES - EXPLAIN “查询” DROP DATABASE “Ownership” DESC DATABASE - CACHE TABLE “查询” UNCACHE TABLE “查询” CLEAR CACHE TABLE “管理” REFRESH TABLE “查询” ADD FILE “管理” ADD JAR “管理” HEALTHCHECK -