华为云用户手册

数据规划 Spark Streaming样例工程的数据存储在Kafka组件中。向Kafka组件发送数据（需要有Kafka权限用户）。 确保集群安装完成，包括HDFS、Yarn、Spark和Kafka。 本地新建文件“input_data1.txt”，将“log1.txt”的内容复制保存到“input_data1.txt”。 在客户端安装节点下创建文件目录：“/home/data”。将上述文件上传到此“/home/data”目录下。 将Kafka的Broker配置参数“allow.everyone.if.no.acl.found”的值修改为“true”。 创建Topic。 {zkQuorum}表示ZooKeeper集群信息，格式为IP:port。 $KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper {zkQuorum}/kafka --replication-factor 1 --partitions 3 --topic {Topic} 启动Kafka的Producer，向Kafka发送数据。 java -cp {ClassPath} com.huawei.bigdata.spark.examples.StreamingExampleProducer {BrokerList} {Topic} 其中，ClassPath应包含Spark客户端Kafka jar包的绝对路径，如/opt/client/Spark2x/spark/jars/*:/opt/client/Spark2x/spark/jars/streamingClient010/*

打包项目 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上。 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 编译打包前，样例代码中的user.keytab、krb5.conf文件路径需要修改为该文件所在客户端服务器的实际路径。例如：“/opt/female/user.keytab”，“/opt/female/krb5.conf”。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“ /opt” ）下。 准备依赖包，将下列jar包上传到Spark客户端所在服务器，“$SPARK_HOME/jars/streamingClient010”目录下。 spark-streaming-kafkaWriter-0-10_2.12-3.1.1-hw-ei-311001.jar kafka-clients-xxx.jar kafka_2.12-xxx.jar spark-sql-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar spark-streaming-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar spark-token-provider-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar 版本号中包含hw-ei的依赖包请从华为开源镜像站下载。 版本号中不包含hw-ei的依赖包都来自开源仓库，请从Maven中心仓获取。

数据规划 准备用户认证文件：登录Manager下载用户凭证，获取“user.keytab”和“krb5.conf”文件。 准备待创建的租户信息，如“tenantId”为“92”，“tenantName”为“test92”，“remark”为“test tenant remark1”。 如果是在Windows运行本样例程序，需将所有FlinkServer所在节点的主机名和IP地址添加到“C:\Windows\System32\drivers\etc\hosts”中。

功能介绍 主要分为三个部分： 从HDFS原文件中抽取name信息，查询HBase、Hive相关数据，并进行数据拼接，通过类MultiComponentMapper继承Mapper抽象类实现。 获取拼接后的数据取最后一条输出到HBase、HDFS，通过类MultiComponentReducer继承Reducer抽象类实现。 main方法提供建立一个MapReduce job，并提交MapReduce作业到Hadoop集群。

操作步骤 参考导入并配置HetuEngine Python3样例工程章节，获取样例代码，获取hetu-jdbc-XXX.jar文件，并复制到自定义目录中。 编辑样例代码，根据集群实际情况修改url、user等信息，并根据实际路径修改“jdbc_location”。 Windows系统路径填写示例："D:\\hetu-examples-python3\\hetu-jdbc-XXX.jar" Linux系统路径示例："/opt/hetu-examples-python3/hetu-jdbc-XXX.jar" 运行python3样例代码。 Windows直接通过pycharm或者Python IDLE运行py脚本。 Linux运行样例代码前需要已安装Java。 进入样例代码路径并执行py脚本，样例代码路径如“/opt/hetu-examples-python3”： cd /opt/hetu-examples-python3 python3 JDBCExampleBroker.py 运行结果。 Windows系统中，在console界面可以看到运行结果示例： 图1 Windows系统运行结果 Linux系统中运行结果示例： … [ ('hive', ('mv',), ('svc',), ('system',), ('systemremote',)]

配置样例代码 在开发环境IntelliJ IDEA中，单击“src/springboot/hbase-examples/src/main/resources”目录下的“springclient.properties”文件，按需修改如下表1中提供的参数： 表1 配置说明表 配置名称 含义 conf.path HBase配置文件所在目录，即“hbase-example\src\main\resources\conf”。

问题现象 Spark能对接很多的第三方工具，因此在使用过程中经常会依赖一堆的三方包。而有一些包 MRS 已经自带，这样就有可能造成代码使用的jar包版本和集群自带的jar包版本不一致，在使用过程中就有可能出现jar包冲突的情况。 常见的jar包冲突报错有： 1、报错类找不到：java.lang.NoClassDefFoundError 2、报错方法找不到：java.lang.NoSuchMethodError

原因分析 以自定义UDF为例： 报错信息显示是找不到类。 首先需要确认的是这个类属于的jar包是否在jvm的classpath里面， spark自带的jar都在“spark客户端目录/jars/”。 确认是否存在多个jar包拥有这个类。 如果是其他依赖包，可能是没有使用--jars添加到任务里面。 如果是已经添加到任务里面，但是依旧没有取到，可能是因为配置文件的driver或者executor的classpath配置不正确，可以查看日志确认是否加载到环境。 另外可能报错是类初始化失败导致后面使用这个类的时候出现上述报错，需要确认是否在之前就有初始化失败或者其他报错的情况发生。 报错信息显示找不到方法。 确认这个方法对应的类所在的jar包是否加载到jvm的classpath里面，spark自带的类都在“spark客户端目录/jars/”。 确认是否有多个jar包包含这个类（尤其注意相同工具的不同版本）。 如果报错是Hadoop相关的包，有可能是因为使用的Hadoop版本不一致导致部分方法已经更改。 如果报错的是三方包里面的类，可能是因为Spark已经自带了相关的jar包，但是和代码中使用的版本不一致。

回答 原因分析：显示该异常是因为“recoverFromCheckpointLocation”的值判定为false，但却配置了checkpoint目录。 参数“recoverFromCheckpointLocation”的值为代码中“outputMode == OutputMode.Complete()”语句的判断结果（outputMode的默认输出方式为“append”）。 处理方法：编写应用时，用户可以根据具体情况修改数据的输出方式。 将输出方式修改为“complete”，“recoverFromCheckpointLocation”的值会判定为true。此时配置了checkpoint目录时就不会显示异常。

问题 Structured Streaming的cluster模式，在数据处理过程中终止ApplicationManager，执行应用时显示如下异常。 2017-05-09 20:46:02,393 | INFO | main | client token: Token { kind: YARN_CLIENT_TOKEN, service: } diagnostics: User class threw exception: org.apache.spark.sql.AnalysisException: This query does not support recovering from checkpoint location. Delete hdfs://hacluster/structuredtest/checkpoint/offsets to start over.; ApplicationMaster host: 10.96.101.170 ApplicationMaster RPC port: 0 queue: default start time: 1494333891969 final status: FAILED tracking URL: https://9-96-101-191:8090/proxy/application_1493689105146_0052/ user: spark2x | org.apache.spark.internal.Logging$class.logInfo(Logging.scala:54) Exception in thread "main" org.apache.spark.SparkException: Application application_1493689105146_0052 finished with failed status

代码样例 由于pyspark不提供Hbase相关api，本样例使用Python调用Java的方式实现。 下面代码片段仅为演示，具体代码参见SparkHbasetoHbasePythonExample： # -*- coding:utf-8 -*- from py4j.java_gateway import java_import from pyspark.sql import SparkSession # 创建SparkSession，设置kryo序列化 spark = SparkSession\ .builder\ .appName("SparkHbasetoHbase") \ .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ .config("spark.kryo.registrator", "com.huawei.bigdata.spark.examples.MyRegistrator") \ .getOrCreate() # 向sc._jvm中导入要运行的类 java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.SparkHbasetoHbase') # 创建类实例并调用方法 spark._jvm.SparkHbasetoHbase().hbasetohbase(spark._jsc) # 停止SparkSession spark.stop()

代码样例 下面代码片段在com.huawei.hadoop.hbase.example包的“HBaseSample”类的testScanDataByIndex方法中： 样例：使用二级索引查找数据 public void testScanDataByIndex() { LOG .info("Entering testScanDataByIndex."); Table table = null; ResultScanner scanner = null; try { table = conn.getTable(tableName); // Create a filter for indexed column. Filter filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"), CompareOperator.EQUAL, "Li Gang".getBytes()); Scan scan = new Scan(); scan.setFilter(filter); scanner = table.getScanner(scan); LOG.info("Scan indexed data."); for (Result result : scanner) { for (Cell cell : result.rawCells()) { LOG.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)), Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell))); } } LOG.info("Scan data by index successfully."); } catch (IOException e) { LOG.error("Scan data by index failed."); } finally { if (scanner != null) { // Close the scanner object. scanner.close(); } try { if (table != null) { table.close(); } } catch (IOException e) { LOG.error("Close table failed."); } } LOG.info("Exiting testScanDataByIndex."); }

功能介绍 针对添加了二级索引的用户表，您可以通过Filter来查询数据。其数据查询性能高于针对无二级索引用户表的数据查询。 HIndex支持的Filter类型为“SingleColumnValueFilter”，“SingleColumnValueExcludeFilter”以及“SingleColumnValuePartitionFilter”。 HIndex支持的Comparator为“BinaryComparator”，“BitComparator”，“LongComparator”，“DecimalComparator”，“DoubleComparator”，“FloatComparator”，“IntComparator”，“NullComparator”。 二级索引的使用规则如下： 针对某一列或者多列创建了单索引的场景下： 当查询时使用此列进行过滤时，不管是AND还是OR操作，该索引都会被利用来提升查询性能。 例如：Filter_Condition(IndexCol1) AND/OR Filter_Condition(IndexCol2) 当查询时使用“索引列AND非索引列”过滤时，此索引会被利用来提升查询性能。 例如：Filter_Condition(IndexCol1) AND Filter_Condition(IndexCol2) AND Filter_Condition(NonIndexCol1) 当查询时使用“索引列OR非索引列”过滤时，此索引将不会被使用，查询性能不会因为索引得到提升。 例如：Filter_Condition(IndexCol1) AND/OR Filter_Condition(IndexCol2) OR Filter_Condition(NonIndexCol1) 针对多个列创建的联合索引场景下： 当查询时使用的列（多个），是联合索引所有对应列的一部分或者全部，且列的顺序与联合索引一致时，此索引会被利用来提升查询性能。 例如，针对C1、C2、C3列创建了联合索引，生效的场景包括： Filter_Condition(IndexCol1) AND Filter_Condition(IndexCol2) AND Filter_Condition(IndexCol3) Filter_Condition(IndexCol1) AND Filter_Condition(IndexCol2) Filter_Condition(IndexCol1) 不生效的场景包括： Filter_Condition(IndexCol2) AND Filter_Condition(IndexCol3) Filter_Condition(IndexCol1) AND Filter_Condition(IndexCol3) Filter_Condition(IndexCol2) Filter_Condition(IndexCol3) 当查询时使用“索引列AND非索引列”过滤时，此索引会被利用来提升查询性能。 例如： Filter_Condition(IndexCol1) AND Filter_Condition(NonIndexCol1) Filter_Condition(IndexCol1) AND Filter_Condition(IndexCol2) AND Filter_Condition(NonIndexCol1) 当查询时使用“索引列OR非索引列”过滤时，此索引不会被使用，查询性能不会因为索引得到提升。 例如： Filter_Condition(IndexCol1) OR Filter_Condition(NonIndexCol1) (Filter_Condition(IndexCol1) AND Filter_Condition(IndexCol2))OR ( Filter_Condition(NonIndexCol1)) 当查询时使用多个列进行范围查询时，只有联合索引中最后一个列可指定取值范围，前面的列只能设置为“=”。 例如：针对C1、C2、C3列创建了联合索引，需要进行范围查询时，只能针对C3设置取值范围，过滤条件为“C1=XXX，C2=XXX，C3=取值范围”。 针对添加了二级索引的用户表，可以通过Filter来查询数据，在单列索引和复合列索引上进行过滤查询，查询结果都与无索引结果相同，且其数据查询性能高于无二级索引用户表的数据查询性能。

回答 由于浏览器所在的计算机IP地址未加到Web访问白名单导致。用户可以通过修改客户端的配置文件“conf/flink-conf.yaml”来解决问题。 确认配置项“jobmanager.web.ssl.enabled”的值是否是“false”，若不是，请修改为“false”。 确认配置项“jobmanager.web.access-control-allow-origin”和“jobmanager.web.allow-access-address”中是否已经添加浏览器所在的计算机IP地址。如果没有添加，可以通过这两项配置项进行添加。例如： jobmanager.web.access-control-allow-origin: 浏览器所在的计算机IP地址 jobmanager.web.allow-access-address: 浏览器所在的计算机IP地址

Oozie Java接口介绍 Java API主要由org.apache.oozie.client.OozieClient提供。 表1 接口介绍 方法 说明 public String run(Properties conf) 运行Job。 public void start(String jobId) 启动指定的Job。 public String submit(Properties conf) 提交Job。 public void kill(String jobId) 删除指定的Job。 public void suspend(String jobId) 暂停指定的Job。 public void resume(String jobId) 恢复指定的Job。 public WorkflowJob getJobInfo(String jobId) 获取Job信息。 父主题： 常用Oozie API接口介绍

准备本地应用开发环境 在进行二次开发时，要准备的开发和运行环境如表1所示。 表1 开发环境 准备项 说明 操作系统 Windows系统，支持Windows 7以上版本。 开发和运行环境需要和集群业务平面网络互通。 安装JDK 开发和运行环境的基本配置。版本要求如下： 服务端和客户端仅支持自带的OpenJDK，版本为1.8.0_272，不允许替换。 对于客户应用需引用SDK类的Jar包运行在客户应用进程中的。 X86客户端：Oracle JDK支持1.8版本；IBM JDK支持1.8.5.11版本。 TaiShan客户端：OpenJDK：支持1.8.0_272版本。 说明： 基于安全考虑，服务端只支持TLS V1.2及以上的加密协议。 安装和配置IDEA 用于开发Oozie应用程序的工具。版本要求：支持JDK1.8以上的版本。 说明： 若使用IBM JDK，请确保IDEA中的JDK配置为IBM JDK。 若使用Oracle JDK，请确保IDEA中的JDK配置为Oracle JDK。 若使用Open JDK，请确保IDEA中的JDK配置为Open JDK。 不同的IntelliJ IDEA不要使用相同的workspace和相同路径下的示例工程。 安装Maven 开发环境的基本配置。用于项目管理，贯穿软件开发生命周期。 7-zip 用于解压“*.zip”和“*.rar”文件。 支持7-zip 16.04版本。 父主题： 准备Oozie应用开发环境

在Linux调测程序 编译并生成Jar包，并将Jar包复制到与依赖库文件夹同级的目录“src/main/resources”下，具体步骤请参考在Linux调测程序。 运行Consumer样例工程的命令如下。 java -cp /opt/client/lib/*:/opt/client/src/main/resources com.huawei.bigdata.kafka.example.Consumer

ClickHouse应用开发流程介绍 开发流程中各阶段的说明如图1所示。 图1 ClickHouse应用程序开发流程 表1 ClickHouse应用开发的流程说明 阶段 说明 参考文档 了解基本概念 在开始开发应用前，需要了解ClickHouse的基本概念。 基本概念 准备开发和运行环境 ClickHouse的应用程序支持多种语言开发，主要为Java语言，推荐使用IntelliJ IDEA工具，请根据指导完成开发环境配置。 准备ClickHouse应用开发和运行环境 根据场景开发工程 提供样例工程，帮助用户快速了解ClickHouse各部件的编程接口。 导入并配置ClickHouse样例工程 运行程序及查询结果 用户可以直接通过运行结果查看应用程序运行情况。 在本地Windows环境中调测ClickHouse应用（MRS 3.3.0之前版本） 在Linux环境中调测ClickHouse应用（MRS 3.3.0之前版本） 父主题： ClickHouse开发指南（普通模式）

回答 首先查看ZooKeeper中/flink_base的目录权限是否为：'world,'anyone: cdrwa；如果不是，请修改/flink_base的目录权限为：'world,'anyone: cdrwa，然后继续根据步骤二排查；如果是，请根据步骤二排查。 由于在Flink配置文件中“high-availability.zookeeper.client.acl”默认为“creator”，即谁创建谁有权限，由于原有用户已经使用ZooKeeper上的/flink_base/flink目录，导致新创建的用户访问不了ZooKeeper上的/flink_base/flink目录。 新用户可以通过以下操作来解决问题。 查看客户端的配置文件“conf/flink-conf.yaml”。 修改配置项“high-availability.zookeeper.path.root”对应的ZooKeeper目录，例如：/flink2。 重新提交任务。

功能简介 HBase通过org.apache.hadoop.hbase.client.Admin对象的createTable方法来创建表，并指定表名、列族名。创建表有两种方式（强烈建议采用预分Region建表方式）： 快速建表，即创建表后整张表只有一个Region，随着数据量的增加会自动分裂成多个Region。 预分Region建表，即创建表时预先分配多个Region，此种方法建表可以提高写入大量数据初期的数据写入速度。 表的列名以及列族名不能包含特殊字符，可以由字母、数字以及下划线组成。

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“HBaseSample”类的testCreateTable方法中。 public void testCreateTable() { LOG.info("Entering testCreateTable."); // Specify the table descriptor. TableDescriptorBuilder htd = TableDescriptorBuilder.newBuilder(tableName);（1） // Set the column family name to info. ColumnFamilyDescriptorBuilder hcd = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"));（2） // Set data encoding methods, HBase provides DIFF,FAST_DIFF,PREFIX hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF); // Set compression methods, HBase provides two default compression // methods:GZ and SNAPPY // GZ has the highest compression rate,but low compression and // decompression effeciency,fit for cold data // SNAPPY has low compression rate, but high compression and // decompression effeciency,fit for hot data. // it is advised to use SNAANPPY hcd.setCompressionType(Compression.Algorithm.SNAPPY);//注[1] htd.setColumnFamily(hcd.build()); （3） Admin admin = null; try { // Instantiate an Admin object. admin = conn.getAdmin(); （4） if (!admin.tableExists(tableName)) { LOG.info("Creating table..."); admin.createTable(htd.build());//注[2] （5） LOG.info(admin.getClusterMetrics().toString()); LOG.info(admin.listNamespaceDescriptors().toString()); LOG.info("Table created successfully."); } else { LOG.warn("table already exists"); } } catch (IOException e) { LOG.error("Create table failed " ,e); } finally { if (admin != null) { try { // Close the Admin object. admin.close(); } catch (IOException e) { LOG.error("Failed to close admin " ,e); } } } LOG.info("Exiting testCreateTable."); }

注意事项 注[1] 可以设置列族的压缩方式，代码片段如下： //设置编码算法，HBase提供了DIFF，FAST_DIFF，PREFIX三种编码算法 hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF); //设置文件压缩方式，HBase默认提供了GZ和SNAPPY两种压缩算法 //其中GZ的压缩率高，但压缩和解压性能低，适用于冷数据 //SNAPPY压缩率低，但压缩解压性能高，适用于热数据 //建议默认开启SNAPPY压缩 hcd.setCompressionType(Compression.Algorithm.SNAPPY); 注[2] 可以通过指定起始和结束RowKey，或者通过RowKey数组预分Region两种方式建表，代码片段如下： // 创建一个预划分region的表 byte[][] splits = new byte[4][]; splits[0] = Bytes.toBytes("A"); splits[1] = Bytes.toBytes("H"); splits[2] = Bytes.toBytes("O"); splits[3] = Bytes.toBytes("U"); admin.createTable(htd, splits);

操作场景 为了运行 FusionInsight MRS产品Flink组件的SpringBoot接口样例代码，需要完成下面的操作。当前支持 GaussDB （DWS）样例工程。 该章节以在Linux环境下开发GaussDB（DWS） SpringBoot方式连接Flink服务的应用程序为例。 执行GaussDB（DWS）样例需提前登录GaussDB（DWS）所在节点创建用于接受数据的空表“test_lzh1”，创建命令如下： create table test_lzh1 (id integer not null);

问题 Structured Streaming的cluster模式，在数据处理过程中终止ApplicationManager，执行应用时显示如下异常。 2017-05-09 20:46:02,393 | INFO | main | client token: Token { kind: YARN_CLIENT_TOKEN, service: } diagnostics: User class threw exception: org.apache.spark.sql.AnalysisException: This query does not support recovering from checkpoint location. Delete hdfs://hacluster/structuredtest/checkpoint/offsets to start over.; ApplicationMaster host: 10.96.101.170 ApplicationMaster RPC port: 0 queue: default start time: 1494333891969 final status: FAILED tracking URL: https://9-96-101-191:8090/proxy/application_1493689105146_0052/ user: spark2x | org.apache.spark.internal.Logging$class.logInfo(Logging.scala:54) Exception in thread "main" org.apache.spark.SparkException: Application application_1493689105146_0052 finished with failed status

回答 原因分析：显示该异常是因为“recoverFromCheckpointLocation”的值判定为false，但却配置了checkpoint目录。 参数“recoverFromCheckpointLocation”的值为代码中“outputMode == OutputMode.Complete()”语句的判断结果（outputMode的默认输出方式为“append”）。 处理方法：编写应用时，用户可以根据具体情况修改数据的输出方式。 将输出方式修改为“complete”，“recoverFromCheckpointLocation”的值会判定为true。此时配置了checkpoint目录时就不会显示异常。

配置ClickHouse连接属性 在ClickhouseJDBCHaDemo、Demo、NativeJDBCHaDemo和Util文件创建connection的样例中设置连接属性，如下样例代码设置socket超时时间为60s。 ClickHouseProperties clickHouseProperties = new ClickHouseProperties(); clickHouseProperties.setSocketTimeout(60000); 如果导入并配置ClickHouse样例工程中的“clickhouse-example.properties”配置文件中“sslUsed”参数配置为“true”时，则需要在ClickhouseJDBCHaDemo、Demo、NativeJDBCHaDemo和Util文件创建connection的样例中设置如下连接属性： clickHouseProperties.setSsl(true); clickHouseProperties.setSslMode("none"); 父主题： 开发ClickHouse应用

使用Python提交Flink普通作业 获取样例工程“flink-examples/pyflink-example/pyflink-kafka”中的“pyflink-kafka.py”和“insertData2kafka.sql”。 参考准备本地应用开发环境将准备好的Python虚拟环境打包，获取“venv.zip”文件。 zip -q -r venv.zip venv/ 以root用户登录主管理节点，将1和2获取的“venv.zip”、“pyflink-kafka.py”和“insertData2kafka.sql”文件上传至客户端环境。 per-job模式：将上述文件上传到“客户端安装目录/Flink/flink”。 yarn-application模式：将上述文件和“flink-connector-kafka-实际版本号.jar”包上传到“客户端安装目录/Flink/flink/yarnship”。 修改“pyflink-kafka.py”中的“specific_jars”路径。 per-job模式：修改为SQL文件的实际路径。如：file:///客户端安装目录/Flink/flink/lib/flink-connector-kafka-实际版本号.jar yarn-application模式：修改为：file://"+os.getcwd()+"/../../../../yarnship/flink-connector-kafka-实际版本号.jar 修改“pyflink-kafka.py”中的“file_path”路径。 per-job模式：修改为sql文件的实际路径。如：客户端安装目录/Flink/flink/insertData2kafka.sql yarn-application模式：修改为os.getcwd() + "/../../../../yarnship/insertData2kafka.sql" 执行以下命令指定运行环境。 export PYFLINK_CLIENT_EXECUTABLE=venv.zip/venv/bin/python3 执行以下命令运行程序。 per-job模式： ./bin/flink run --detached -t yarn-per-job -Dyarn.application.name=py_kafka -pyarch venv.zip -pyexec venv.zip/venv/bin/python3 -py pyflink-kafka.py 运行结果： yarn-application模式 ./bin/flink run-application --detached -t yarn-application -Dyarn.application.name=py_kafka -Dyarn.ship-files=/opt/client/Flink/flink/yarnship/ -pyarch yarnship/venv.zip -pyexec venv.zip/venv/bin/python3 -pyclientexec venv.zip/venv/bin/python3 -pyfs yarnship -pym pyflink-kafka 运行结果： 父主题： PyFlink样例程序

开发思路 写HBase： 通过参数指定“hbase-site.xml”文件的父目录，Flink Sink可以获取到HBase的Connection。 通过Connection判断表是否存在，如果不存在则创建表。 将接收到的数据转化成Put对象，写到HBase。 读HBase： 通过参数指定“hbase-site.xml”文件的父目录，Flink Source可以获取到HBase的Connection。 通过Connection判断表是否存在，如果不存在则作业失败，需要通过HBase Shell创建表或上游作业创建表。 读取HBase中的数据，将Result数据转化成Row对象发送给下游算子。

数据规划 确保以多主实例模式启动了JD BCS erver服务，并至少有一个实例可连接客户端。在JDB CS erver节点上分别创建“/home/data”文件，内容如下： Miranda,32 Karlie,23 Candice,27 确保其对启动JDBCServer的用户有读写权限。 确保客户端classpath下有“hive-site.xml”文件，且根据实际集群情况配置所需要的参数。JDBCServer相关参数详情，请参见Spark JDBCServer接口介绍。

运行任务 进入Spark客户端目录，使用java -cp命令运行代码（类名与文件名等请与实际代码保持一致，此处仅为示例）： 运行Java样例代码： java -cp $SPARK_HOME/jars/*:$SPARK_HOME/jars/hive/*:$SPARK_HOME/conf:/opt/female/SparkThriftServerJavaExample-1.0.jar com.huawei.bigdata.spark.examples.ThriftServerQueriesTest $SPARK_HOME/conf/hive-site.xml $SPARK_HOME/conf/spark-defaults.conf 运行Scala样例代码： java -cp $SPARK_HOME/jars/*:$SPARK_HOME/jars/hive/*:$SPARK_HOME/conf:/opt/female/SparkThriftServerExample-1.0.jar com.huawei.bigdata.spark.examples.ThriftServerQueriesTest $SPARK_HOME/conf/hive-site.xml $SPARK_HOME/conf/spark-defaults.conf 集群开启ZooKeeper的SSL特性后（查看ZooKeeper服务的ssl.enabled参数），请在执行命令中添加-Dzookeeper.client.secure=true -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty两项参数： java -Dzookeeper.client.secure=true -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty -cp $SPARK_HOME/jars/*:$SPARK_HOME/jars/hive/*:$SPARK_HOME/conf:/opt/female/SparkThriftServerJavaExample-1.0.jar com.huawei.bigdata.spark.examples.ThriftServerQueriesTest $SPARK_HOME/conf/hive-site.xml $SPARK_HOME/conf/spark-defaults.conf