问题一：Flink 中 shuffle Partitioner 和 rebalance partitoner 有什么区别？
问题如上，辛苦大神解答下*来自志愿者整理的flink邮件归档
参考回答：
ShufflePartitioner:
public int selectChannel(SerializationDelegate record) { return random.nextInt(numberOfChannels); }
RebalancePartitioner public int selectChannel(SerializationDelegate record) { nextChannelToSendTo = (nextChannelToSendTo + 1) % numberOfChannels; return nextChannelToSendTo; }
一个随机，一个严格 round-robin。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359357?spm=a2c6h.13262185.0.0.54e839c0D2mgIx
问题二：关于pyflink LATERAL TABLE 问题请教
定制UDTF想要拆分字符串、但是报错 java.lang.IndexOutOfBoundsException: Index: 7, Size: 7 毫无头绪、有大佬遇到过吗？谢谢！
class myKerasMLP(ScalarFunction):
def eval(self, *args): ...
返回预测结果
return str(trueY[0][0]) + '|' + str(trueY[0][1])
注册UDF函数 myKerasMLP = udf(myKerasMLP(), input_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING()], result_type=DataTypes.STRING())
class SplitStr(TableFunction): def eval(self, str_value): str_arr = str_value.split('|') yield str_arr[0], str_arr[1] yield str_arr[0], str_arr[1]
注册UDTF函数 splitStr = udtf(SplitStr(), DataTypes.STRING(), [DataTypes.STRING(), DataTypes.STRING()])
t_env.register_function('train_and_predict', myKerasMLP) t_env.register_function("splitStr", splitStr)
==================
t_env.sql_query(""" select A.hotime , A.before_ta , A.before_rssi , A.after_ta , A.after_rssil , A.nb_tath , A.nb_rssith , nbr_rssi , nbr_ta from (SELECT hotime , before_ta , before_rssi , after_ta , after_rssil , nb_tath , nb_rssith , train_and_predict(hotime, before_ta, before_rssi, after_ta, after_rssil, nb_tath, nb_rssith) predict FROM source) as A, LATERAL TABLE(splitStr(predict)) as T(nbr_rssi, nbr_ta) """).insert_into("predict_sink")
==================== 报错 java.lang.IndexOutOfBoundsException: Index: 7, Size: 7 Traceback (most recent call last): File "C:/Users/Administrator.XTZ-02012091146/PycharmProjects/pythonProject/kerasTest/UdtfNtPredictPyFlink.py", line 280, in t_env.execute('NT重连预测参数') File "D:\tools\Python3.6.5\lib\site-packages\pyflink\table\table_environment.py", line 1057, in execute return JobExecutionResult(self._j_tenv.execute(job_name)) File "D:\tools\Python3.6.5\lib\site-packages\py4j\java_gateway.py", line 1286, in call answer, self.gateway_client, self.target_id, self.name) File "D:\tools\Python3.6.5\lib\site-packages\pyflink\util\exceptions.py", line 147, in deco return f(*a, **kw) File "D:\tools\Python3.6.5\lib\site-packages\py4j\protocol.py", line 328, in get_return_value format(target_id, ".", name), value) py4j.protocol.Py4JJavaError: An error occurred while calling o25.execute. : java.lang.IndexOutOfBoundsException: Index: 7, Size: 7 at java.util.ArrayList.rangeCheck(ArrayList.java:657)
==================== 这段SQL可以执行 t_env.sql_query(""" SELECT hotime , before_ta , before_rssi , after_ta , after_rssil , nb_tath , nb_rssith , train_and_predict(hotime, before_ta, before_rssi, after_ta, after_rssil, nb_tath, nb_rssith) predict FROM source """).insert_into("print_table")
+I(37,14,-66,92,-74,24,-65,22.621065|-64.12096) +I(291,136,-76,136,-78,22,-65,19.479145|-65.958) ------------------------------*来自志愿者整理的flink邮件归档
参考回答：
apache-flink 1.11.1
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359385?spm=a2c6h.13262185.0.0.54e839c0LrOat6
问题三：1.12 yarn-per-job提交作业失败怎么办？
> 我在用这个命令提交的时候会报
> flink Application rejected by queue placement policy
> 这个应该是没有指定queue
> 但是我在命令中加了-yqu 这个参数，在web界面看quene 的时候，不是我指定的，而是default 。
> 另外，我用旧命令提交作业可以正常运行。请问有人碰到过这个问题吗。*来自志愿者整理的flink邮件归档
参考回答：
从 Flink 1.12 开始，-yqu 等 YARN 相关的参数被移除了，可以使用 [1] 来代替。
[1 ]https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/config.html#yarn-application-queue
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359388?spm=a2c6h.13262185.0.0.54e839c0LrOat6
问题四：Pyflink如何对接HBase？
请问pyflink如何对接hbase，有没有案例？*来自志愿者整理的flink邮件归档
参考回答：
1）在PyFlink Table API中可以使用所有SQL中支持的connector，所以HBase connector也自然支持，具体使用方式可以看一下文档： https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/python_table_api_connectors.html#how-to-use-connectors 2）HBase connector的使用方式可以看一下： https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/connectors/hbase.html
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359389?spm=a2c6h.13262185.0.0.54e839c0ROdVpg
问题五：PyFlink UDTF 运行一段时间后报 NullPointerException怎么办？
PyFlink UDTF 运行一段时间后报 NullPointerException怎么办？
参考回答：
这个问题应该和这个JIRA有关系：https://issues.apache.org/jira/browse/FLINK-21434
目前已经在master和1.12.3上修复了，但是1.12.3还没有release。你要不cherry-pick这个fix，自己build一个版本，验证一下？
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359392?spm=a2c6h.13262185.0.0.54e839c0ROdVpg
