NameError: name 'ScalarFunction' is not defined
你 import ScalarFunction了吗？
On Tue, Apr 19, 2022 at 2:08 PM 799590...@qq.com.INVALID
<799590...@qq.com.invalid> wrote:
>
> 以下是刚刚的报错日志，现在里面没有Python callback server start
> failed了，是因为原来黄底部分获取的Python环境变量的路径为C:\Python310（早先安装的python,后来安装python36，设置环境PATH变量为python36），重启机器后好了
>
> 2022-04-19 13:58:11 |INFO |http-nio-9532-exec-3 |HiveCatalog.java:257
> |org.apache.flink.table.catalog.hive.HiveCatalog |Setting hive conf dir as
> /home/tetris/conf
> 2022-04-19 13:58:11 |INFO |http-nio-9532-exec-3 |HiveCatalog.java:219
> |org.apache.flink.table.catalog.hive.HiveCatalog |Created HiveCatalog
> 'myhive'
> 2022-04-19 13:58:11 |INFO |http-nio-9532-exec-3 |HiveCatalog.java:299
> |org.apache.flink.table.catalog.hive.HiveCatalog |Connected to Hive
> metastore
> 2022-04-19 13:58:11 |INFO |http-nio-9532-exec-3 |CatalogManager.java:262
> |org.apache.flink.table.catalog.CatalogManager |Set the current default
> catalog as [myhive] and the current default database as [tetris].
> 2022-04-19 13:58:11 |INFO |http-nio-9532-exec-3
> |SqlTaskRunService.java:599
> |com.chinaoly.tetris.flink.service.SqlTaskRunService |开始加载hdfs上的udf
> 2022-04-19 13:58:12 |INFO |http-nio-9532-exec-3 |EnvUtil.java:31
> |com.chinaoly.tetris.flink.util.EnvUtil |registerFactory :
> org.apache.hadoop.fs.FsUrlStreamHandlerFactory
> 2022-04-19 13:58:12 |INFO |http-nio-9532-exec-3 |EnvUtil.java:55
> |com.chinaoly.tetris.flink.util.EnvUtil
> |hdfs://chinaoly/tetris/file/function/flink/myReplace.py
> 2022-04-19 13:58:12 |INFO |http-nio-9532-exec-3
> |SqlTaskRunService.java:621
> |com.chinaoly.tetris.flink.service.SqlTaskRunService |完成加载hdfs上的udf
> 13:58:20,956 |-INFO in ReconfigureOnChangeTask(born:1650344954461) - Empty
> watch file list. Disabling
> 2022-04-19 13:58:21 |INFO |http-nio-9532-exec-3 |PythonEnvUtils.java:284
> |org.apache.flink.client.python.PythonEnvUtils |Starting Python process
> with environment variables: {USERDOMAIN_ROAMINGPROFILE=DESKTOP-LBP3EGP,
> HADOOP_CONF_DIR=/home/tetris/conf, PROCESSOR_LEVEL=6, SESSIONNAME=Console,
> ALLUSERSPROFILE=C:\ProgramData, PROCESSOR_ARCHITECTURE=AMD64,
> PSModulePath=C:\Program
> Files\WindowsPowerShell\Modules;C:\WINDOWS\system32\WindowsPowerShell\v1.0\Modules,
> SystemDrive=C:, MAVEN_HOME=E:\Program Files\apache-maven-3.8.2,
> USERNAME=Administrator, ProgramFiles(x86)=C:\Program Files (x86),
> FPS_BROWSER_USER_PROFILE_STRING=Default,
> PYTHONPATH=C:/Users/ADMINI~1/AppData/Local/Temp/pyflink/df5887da-8c4d-4807-b891-c313338f1c14/63150f6e-5ea0-4186-be56-ce25e69c4265,
> PATHEXT=.COM;.EXE;.BAT;.CMD;.VBS;.VBE;.JS;.JSE;.WSF;.WSH;.MSC,
> DriverData=C:\Windows\System32\Drivers\DriverData, PYTHON_HOME=E:\Program
> Files\python36, ProgramData=C:\ProgramData, ProgramW6432=C:\Program Files,
> HOMEPATH=\Users\Administrator, PROCESSOR_IDENTIFIER=Intel64 Family 6 Model
> 94 Stepping 3, GenuineIntel, ProgramFiles=C:\Program Files,
> PUBLIC=C:\Users\Public, windir=C:\WINDOWS, =::=::\, ZES_ENABLE_SYSMAN=1,
> LOCALAPPDATA=C:\Users\Administrator\AppData\Local,
> ChocolateyLastPathUpdate=132792636909481380, IntelliJ IDEA=E:\Program
> Files\JetBrains\IntelliJ IDEA 2019.3.3\bin;, USERDOMAIN=DESKTOP-LBP3EGP,
> FPS_BROWSER_APP_PROFILE_STRING=Internet Explorer,
> LOGONSERVER=\\DESKTOP-LBP3EGP, JAVA_HOME=C:\Program
> Files\Java\jdk1.8.0_261, OneDrive=C:\Users\Administrator\OneDrive,
> APPDATA=C:\Users\Administrator\AppData\Roaming,
> ChocolateyInstall=C:\ProgramData\chocolatey, SCALA_HOME=C:\Program Files
> (x86)\scala, CommonProgramFiles=C:\Program Files\Common Files,
> Path=C:\Program Files (x86)\Common
> Files\Oracle\Java\javapath;C:\WINDOWS\system32;C:\WINDOWS;C:\WINDOWS\System32\Wbem;C:\WINDOWS\System32\WindowsPowerShell\v1.0\;C:\WINDOWS\System32\OpenSSH\;C:\Program
> Files\Git\cmd;C:\Program Files\Java\jdk1.8.0_261\bin;E:\Program
> Files\apache-maven-3.8.2\bin;E:\Program
> Files\TortoiseSVN\bin;%HADOOP_HOME%\bin;E:\Program
> Files\nodejs\;C:\ProgramData\chocolatey\bin;C:\Program Files
> (x86)\scala\bin;C:\Program Files (x86)\scala\bin;E:\Program
> Files\python36;E:\Program Files\python36\Scripts\;E:\Program
> Files\python36\;C:\Users\Administrator\AppData\Local\Microsoft\WindowsApps;;E:\Program
> Files\JetBrains\IntelliJ IDEA
> 2019.3.3\bin;;C:\Users\Administrator\AppData\Roaming\npm,
> JETBRAINS_LICENSE_SERVER=http://fls.jetbrains-agent.com, OS=Windows_NT,
> COMPUTERNAME=DESKTOP-LBP3EGP, PROCESSOR_REVISION=5e03,
> CommonProgramW6432=C:\Program Files\Common Files,
> PYFLINK_GATEWAY_PORT=53944, ComSpec=C:\WINDOWS\system32\cmd.exe,
> SystemRoot=C:\WINDOWS, TEMP=C:\Users\ADMINI~1\AppData\Local\Temp,
> HOMEDRIVE=C:, USERPROFILE=C:\Users\Administrator,
> TMP=C:\Users\ADMINI~1\AppData\Local\Temp,
> CommonProgramFiles(x86)=C:\Program Files (x86)\Common Files,
> NUMBER_OF_PROCESSORS=4,
> IDEA_INITIAL_DIRECTORY=C:\Users\Administrator\Desktop}, command: python.exe
> -m pyflink.pyflink_callback_server
> 2022-04-19 13:58:22 |ERROR |http-nio-9532-exec-3 |ExceptionHandle.java:31
> |com.chinaoly.frm.core.aop.ExceptionHandle
> |org.apache.flink.table.api.ValidationException: SQL validation failed.
> Cannot instantiate user-defined function 'myhive.tetris.myReplace'.
> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
> $apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:156)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:107)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:205)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:603)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:272)
> at
> org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:101)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:724)
> at
> com.chinaoly.tetris.flink.service.SqlTaskRunService.executeDefaultSql(SqlTaskRunService.java:117)
> at
> com.chinaoly.tetris.flink.controller.SqlTaskRunController.executeDefaultSql(SqlTaskRunController.java:66)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at
> org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:209)
> at
> org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:136)
> at
> org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:102)
> at
> org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:893)
> at
> org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:799)
> at
> org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
> at
> org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:991)
> at
> org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:925)
> at
> org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:981)
> at
> org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:884)
> at javax.servlet.http.HttpServlet.service(HttpServlet.java:661)
> at
> org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:858)
> at javax.servlet.http.HttpServlet.service(HttpServlet.java:742)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:52)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.CorsFilter.doFilterInternal(CorsFilter.java:96)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> com.alibaba.druid.support.http.WebStatFilter.doFilter(WebStatFilter.java:124)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.HttpPutFormContentFilter.doFilterInternal(HttpPutFormContentFilter.java:109)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:93)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:200)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:199)
> at
> org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96)
> at
> org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:493)
> at
> org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:137)
> at
> org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:81)
> at
> org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:87)
> at
> org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343)
> at
> org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:798)
> at
> org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66)
> at
> org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:806)
> at org.apache.tomcat.util.net
> .NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1498)
> at org.apache.tomcat.util.net
> .SocketProcessorBase.run(SocketProcessorBase.java:49)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at
> org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: org.apache.flink.table.api.ValidationException: Cannot
> instantiate user-defined function 'myhive.tetris.myReplace'.
> at
> org.apache.flink.table.functions.UserDefinedFunctionHelper.instantiateFunction(UserDefinedFunctionHelper.java:219)
> at
> org.apache.flink.table.catalog.FunctionCatalog.getFunctionDefinition(FunctionCatalog.java:659)
> at
> org.apache.flink.table.catalog.FunctionCatalog.resolvePreciseFunctionReference(FunctionCatalog.java:582)
> at
> org.apache.flink.table.catalog.FunctionCatalog.lambda$resolveAmbiguousFunctionReference$3(FunctionCatalog.java:624)
> at java.util.Optional.orElseGet(Optional.java:267)
> at
> org.apache.flink.table.catalog.FunctionCatalog.resolveAmbiguousFunctionReference(FunctionCatalog.java:624)
> at
> org.apache.flink.table.catalog.FunctionCatalog.lookupFunction(FunctionCatalog.java:362)
> at
> org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:97)
> at
> org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:67)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1182)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1168)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1199)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1168)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1168)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:944)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:703)
> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
> $apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:151)
> ... 65 more
> Caused by: java.lang.IllegalStateException: Instantiating python function
> 'myReplace.MyReplace' failed.
> at
> org.apache.flink.table.functions.python.utils.PythonFunctionUtils.getPythonFunction(PythonFunctionUtils.java:48)
> at
> org.apache.flink.table.functions.UserDefinedFunctionHelper.instantiateFunction(UserDefinedFunctionHelper.java:206)
> ... 81 more
> Caused by: java.lang.reflect.InvocationTargetException
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at
> org.apache.flink.table.functions.python.utils.PythonFunctionUtils.getPythonFunction(PythonFunctionUtils.java:45)
> ... 82 more
> Caused by: org.apache.flink.api.python.shaded.py4j.Py4JException: An
> exception was raised by the Python Proxy. Return Message: Traceback (most
> recent call last):
> File "E:\Program Files\python36\lib\site-packages\py4j\java_gateway.py",
> line 2410, in _call_proxy
> return_value = getattr(self.pool[obj_id], method)(*params)
> File "E:\Program
> Files\python36\lib\site-packages\pyflink\java_gateway.py", line 169, in
> getPythonFunction
> udf_wrapper = getattr(importlib.import_module(moduleName), objectName)
> File "E:\Program Files\python36\lib\importlib\__init__.py", line 126, in
> import_module
> return _bootstrap._gcd_import(name[level:], package, level)
> File "<frozen importlib._bootstrap>", line 994, in _gcd_import
> File "<frozen importlib._bootstrap>", line 971, in _find_and_load
> File "<frozen importlib._bootstrap>", line 955, in
> _find_and_load_unlocked
> File "<frozen importlib._bootstrap>", line 665, in _load_unlocked
> File "<frozen importlib._bootstrap_external>", line 678, in exec_module
> File "<frozen importlib._bootstrap>", line 219, in
> _call_with_frames_removed
> File
> "C:\Users\ADMINI~1\AppData\Local\Temp\pyflink\df5887da-8c4d-4807-b891-c313338f1c14\63150f6e-5ea0-4186-be56-ce25e69c4265\myReplace.py",
> line 3, in <module>
> class MyReplace(ScalarFunction):
> NameError: name 'ScalarFunction' is not defined
>
> at
> org.apache.flink.api.python.shaded.py4j.Protocol.getReturnValue(Protocol.java:476)
> at
> org.apache.flink.api.python.shaded.py4j.reflection.PythonProxyHandler.invoke(PythonProxyHandler.java:108)
> at com.sun.proxy.$Proxy159.getPythonFunction(Unknown Source)
> at
> org.apache.flink.client.python.PythonFunctionFactoryImpl.getPythonFunction(PythonFunctionFactoryImpl.java:47)
> at
> org.apache.flink.client.python.PythonFunctionFactory.getPythonFunction(PythonFunctionFactory.java:131)
> ... 87 more
>
> 2022-04-19 13:58:22 |ERROR |http-nio-9532-exec-3 |DirectJDKLog.java:182
> |org.apache.catalina.core.ContainerBase.[Tomcat].[localhost].[/].[dispatcherServlet]
> |Servlet.service() for servlet [dispatcherServlet] in context with path []
> threw exception [Request processing failed; nested exception is
> org.apache.flink.table.api.ValidationException: SQL validation failed.
> Cannot instantiate user-defined function 'myhive.tetris.myReplace'.] with
> root cause
> org.apache.flink.api.python.shaded.py4j.Py4JException: An exception was
> raised by the Python Proxy. Return Message: Traceback (most recent call
> last):
> File "E:\Program Files\python36\lib\site-packages\py4j\java_gateway.py",
> line 2410, in _call_proxy
> return_value = getattr(self.pool[obj_id], method)(*params)
> File "E:\Program
> Files\python36\lib\site-packages\pyflink\java_gateway.py", line 169, in
> getPythonFunction
> udf_wrapper = getattr(importlib.import_module(moduleName), objectName)
> File "E:\Program Files\python36\lib\importlib\__init__.py", line 126, in
> import_module
> return _bootstrap._gcd_import(name[level:], package, level)
> File "<frozen importlib._bootstrap>", line 994, in _gcd_import
> File "<frozen importlib._bootstrap>", line 971, in _find_and_load
> File "<frozen importlib._bootstrap>", line 955, in
> _find_and_load_unlocked
> File "<frozen importlib._bootstrap>", line 665, in _load_unlocked
> File "<frozen importlib._bootstrap_external>", line 678, in exec_module
> File "<frozen importlib._bootstrap>", line 219, in
> _call_with_frames_removed
> File
> "C:\Users\ADMINI~1\AppData\Local\Temp\pyflink\df5887da-8c4d-4807-b891-c313338f1c14\63150f6e-5ea0-4186-be56-ce25e69c4265\myReplace.py",
> line 3, in <module>
> class MyReplace(ScalarFunction):
> NameError: name 'ScalarFunction' is not defined
> at
> org.apache.flink.api.python.shaded.py4j.Protocol.getReturnValue(Protocol.java:476)
> at
> org.apache.flink.api.python.shaded.py4j.reflection.PythonProxyHandler.invoke(PythonProxyHandler.java:108)
> at com.sun.proxy.$Proxy159.getPythonFunction(Unknown Source)
> at
> org.apache.flink.client.python.PythonFunctionFactoryImpl.getPythonFunction(PythonFunctionFactoryImpl.java:47)
> at
> org.apache.flink.client.python.PythonFunctionFactory.getPythonFunction(PythonFunctionFactory.java:131)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at
> org.apache.flink.table.functions.python.utils.PythonFunctionUtils.getPythonFunction(PythonFunctionUtils.java:45)
> at
> org.apache.flink.table.functions.UserDefinedFunctionHelper.instantiateFunction(UserDefinedFunctionHelper.java:206)
> at
> org.apache.flink.table.catalog.FunctionCatalog.getFunctionDefinition(FunctionCatalog.java:659)
> at
> org.apache.flink.table.catalog.FunctionCatalog.resolvePreciseFunctionReference(FunctionCatalog.java:582)
> at
> org.apache.flink.table.catalog.FunctionCatalog.lambda$resolveAmbiguousFunctionReference$3(FunctionCatalog.java:624)
> at java.util.Optional.orElseGet(Optional.java:267)
> at
> org.apache.flink.table.catalog.FunctionCatalog.resolveAmbiguousFunctionReference(FunctionCatalog.java:624)
> at
> org.apache.flink.table.catalog.FunctionCatalog.lookupFunction(FunctionCatalog.java:362)
> at
> org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:97)
> at
> org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:67)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1182)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1168)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1199)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1168)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1168)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:944)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:703)
> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
> $apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:151)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:107)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:205)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:603)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:272)
> at
> org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:101)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:724)
> at
> com.chinaoly.tetris.flink.service.SqlTaskRunService.executeDefaultSql(SqlTaskRunService.java:117)
> at
> com.chinaoly.tetris.flink.controller.SqlTaskRunController.executeDefaultSql(SqlTaskRunController.java:66)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at
> org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:209)
> at
> org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:136)
> at
> org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:102)
> at
> org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:893)
> at
> org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:799)
> at
> org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
> at
> org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:991)
> at
> org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:925)
> at
> org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:981)
> at
> org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:884)
> at javax.servlet.http.HttpServlet.service(HttpServlet.java:661)
> at
> org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:858)
> at javax.servlet.http.HttpServlet.service(HttpServlet.java:742)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:52)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.CorsFilter.doFilterInternal(CorsFilter.java:96)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> com.alibaba.druid.support.http.WebStatFilter.doFilter(WebStatFilter.java:124)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.HttpPutFormContentFilter.doFilterInternal(HttpPutFormContentFilter.java:109)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:93)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:200)
> at
> org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
> at
> org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
> at
> org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
> at
> org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:199)
> at
> org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96)
> at
> org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:493)
> at
> org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:137)
> at
> org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:81)
> at
> org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:87)
> at
> org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343)
> at
> org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:798)
> at
> org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66)
> at
> org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:806)
> at org.apache.tomcat.util.net
> .NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1498)
> at org.apache.tomcat.util.net
> .SocketProcessorBase.run(SocketProcessorBase.java:49)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at
> org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
> at java.lang.Thread.run(Thread.java:748)
>
>
>
> 799590...@qq.com
>
> From: Dian Fu
> Date: 2022-04-19 13:49
> To: user-zh; 799590989
> Subject: Re: Python callback server start failed
> 发一下完整的日志文件？
>
> On Tue, Apr 19, 2022 at 10:53 AM 799590...@qq.com.INVALID
> <799590...@qq.com.invalid> wrote:
>
> > 软件版本
> >
> > flink-1.13.6
> > python3.6.8
> > 本地win10也安装了python3.6.8且添加了python的环境变量和成功运行了$ python -m pip install
> > apache-flink==1.13.6
> > standalonesession方式部署的，一个JM 两个TM，3台集群都安装了python3.6.8 且安装了pyflink-1.13.6
> >
> > 问题：
> >
> > 1、调用python udf时会报如下错误
> > Servlet.service() for servlet [dispatcherServlet] in context with path []
> > threw exception [Request processing failed; nested exception is
> > org.apache.flink.table.api.ValidationException: SQL validation failed.
> > Cannot instantiate user-defined function 'myhive.tetris.myreplace'.] with
> > root cause
> > java.lang.RuntimeException: Python callback server start failed!
> >
> > 2、sql中CREATE FUNCTION 中的 AS 后面的类路径在python环境下的填写有什么规则吗？
> >
> > python udf文件myReplace.py的内容
> >
> > from pyflink.table.expressions import call
> >
> > class MyReplace(ScalarFunction):
> > def __init__(self):
> > self.factor = 12
> >
> > def eval(self, s):
> > return s.replace('buy','sale')
> >
> > 获取远程集群环境，其中的catalogName=myhive，defaultDatabase=tetris
> >
> > StreamExecutionEnvironment env =
> > StreamExecutionEnvironment.createRemoteEnvironment(host,port);
> > env.setStateBackend(new HashMapStateBackend());
> > env.enableCheckpointing(1000);
> > env.getCheckpointConfig().setCheckpointStorage(new
> > JobManagerCheckpointStorage());
> >
> >
> env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
> > env.getCheckpointConfig().setCheckpointTimeout(60000);
> > env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
> > env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
> > env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);
> > env.getCheckpointConfig().setExternalizedCheckpointCleanup(
> > CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION
> > );
> > EnvironmentSettings bsSettings =
> >
> EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
> > StreamTableEnvironment tableEnv =
> > StreamTableEnvironment.create(env,bsSettings);
> >
> >
> tableEnv.getConfig().getConfiguration().setBoolean("table.exec.hive.fallback-mapred-reader",true);
> >
> >
> tableEnv.getConfig().getConfiguration().setBoolean("table.dynamic-table-options.enabled",true);
> > if (!catalogName.equals(tableEnv.getCurrentCatalog())) {
> > HiveCatalog hiveCatalog = new HiveCatalog(catalogName,
> > defaultDatabase, hiveConfDir);
> > tableEnv.registerCatalog(catalogName, hiveCatalog);
> > }
> > tableEnv.useCatalog(catalogName);
> > tableEnv.useDatabase(defaultDatabase);
> > List<String> jars = new ArrayList<>();
> > List<String> pys = new ArrayList<>();
> > log.info("开始加载hdfs上的udf");
> > String prefix = "/file/function/flink/";
> > try {
> > EnvUtil.registerFactory(new FsUrlStreamHandlerFactory());
> > for (String hdf : EnvUtil.listHdfs(prefix,configuration)) {
> > if (hdf.endsWith(".jar")) {
> > jars.add(hdf);
> > EnvUtil.loadJar(URLUtil.url(hdf));
> > } else if (hdf.endsWith(".py")){
> > pys.add(hdf);
> > }
> > }
> >
> >
> tableEnv.getConfig().getConfiguration().set(PipelineOptions.CLASSPATHS,jars);
> >
> >
> tableEnv.getConfig().getConfiguration().setString("python.files",StringUtils.join(pys,","));
> > log.info("完成加载hdfs上的udf");
> > }catch (Exception e){
> > e.printStackTrace();
> > }
> >
> > python文件存放在hdfs指定的路劲下面
> >
> > 上传py文件后通过tableEnv.executeSql 执行了
> >
> > CREATE FUNCTION IF NOT EXISTS myhive.tetris.myReplaceAS
> > 'myReplace.MyReplace' LANGUAGE PYTHON
> >
> > 先行感谢flink官方同学的辛苦付出
> >
> >
> >
> > 799590...@qq.com
> >
>
