Re: flink1.12错误OSError: Expected IPC message of type schema but got record batch

[咿咿呀呀 <20...@qq.com>]

社区的各位大神，有没有碰到这个问题的，请教。



--
Sent from: http://apache-flink.147419.n8.nabble.com/

Re: flink1.12错误OSError: Expected IPC message of type schema but got record batch

[小学生 <20...@qq.com>]

谢谢大佬的指点，我去增大试试

回复： flink1.12错误OSError: Expected IPC message of type schema but got record batch

[刘海 <li...@163.com>]

这个应该和内存有关，我之前试过，存储的状态无限增长，导致运行几分钟后任务结束，并抛出异常，可以尝试一下加大内存和清理状态


| |
刘海
|
|
liuhai35@163.com
|
签名由网易邮箱大师定制
在2021年1月4日 11:35，咿咿呀呀<20...@qq.com> 写道：
我按照您这个修改了，跟我之前的也是一样的。能运行的通，输出的结果也是正确的，现在最大的问题是——运行一段时间后（3分钟左右）就出现了OSError:
Expected IPC message of type schema but got record batch这个错误



--
Sent from: http://apache-flink.147419.n8.nabble.com/

Re: flink1.12错误OSError: Expected IPC message of type schema but got record batch

[咿咿呀呀 <20...@qq.com>]

我按照您这个修改了，跟我之前的也是一样的。能运行的通，输出的结果也是正确的，现在最大的问题是——运行一段时间后（3分钟左右）就出现了OSError:
Expected IPC message of type schema but got record batch这个错误



--
Sent from: http://apache-flink.147419.n8.nabble.com/

Re: flink1.12错误OSError: Expected IPC message of type schema but got record batch

[小学生 <20...@qq.com>]

好的，谢谢Xingbo大佬的指导，我先去试试。

Re: flink1.12错误OSError: Expected IPC message of type schema but got record batch

[Xingbo Huang <hx...@gmail.com>]

Hi,
不好意思，这么晚才回复。因为你这个报错是发生在数据反序列的过程中，还没有到你写的函数体的具体内容。我看你pandas
udf的声明没有问题，那就得看下你的如何使用的了。我写了一个简化版本的，Array[String]作为输入，并且作为输出的，运行起来没有问题。
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings,
DataTypes
from pyflink.table.udf import udf

def extract_data():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)
    t_env = StreamTableEnvironment.create(env,

environment_settings=EnvironmentSettings.new_instance()

.in_streaming_mode().use_blink_planner().build())

    @udf(result_type=DataTypes.ARRAY(DataTypes.STRING()),
func_type="pandas")
    def get_string_element(my_list):
        return my_list

    t = t_env.from_elements(
        [("1", ["1", "2", "3"]), ("3", ["2", "3", "3"]), ("2", ["1", "4",
"3"])],
        DataTypes.ROW(
            [DataTypes.FIELD("Key", DataTypes.STRING()),
             DataTypes.FIELD("List_element",
DataTypes.ARRAY(DataTypes.STRING()))]))
    print(t.select(get_string_element(t.List_element)).to_pandas())


if __name__ == '__main__':
    extract_data()

你可以看下是不是类似的用法，或者你运行我这个demo是不是也会报错。

Best,
Xingbo


咿咿呀呀 <20...@qq.com> 于2021年1月4日周一 上午9:38写道：

> 社区的各位大神，有没有碰到这个问题的，请教。
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/