不能运行基本的 PyFlink 例子

Question

我有这个玩具管道

from pyflink.datastream import StreamExecutionEnvironment


def pipeline():
    # Create environment
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)

    ds = env.read_text_file('file:///home/user/myfile.json')
    # ds.map(lambda i: i)
    ds.print()

    # Execute job
    env.execute('DynamicStockpilePipeline')


if __name__ == '__main__':
    pipeline()

那运行很好，但每次我尝试取消注释映射阶段时，无论是虚拟内联 lambda 还是 MapFunction，它都会爆炸说：

Caused by: java.io.IOException: Failed to execute the command: python
-c import pyflink;import os;print(os.path.join(os.path.abspath(os.path.dirname(pyflink.__file__)), 'bin')) output: Traceback (most recent call last):   File "<string>",line 1, in <module> ModuleNotFoundError: No module named 'pyflink'

我正在使用 pyenv 3.8 解释器，有人知道基本数据源和输出运行是怎么来的，但是地图上说缺少 pyflink 模块吗？

附录：这只发生在 PyCharm 上，当我运行控制台上的脚本时不会发生，所以我担心它是 PyCharm 和 pyenv 之间的东西

Answer 1

或许可以post报错时的代码

根据上下文，这里报错可能是使用错误，print函数需要跟在map函数后面

from pyflink.datastream import StreamExecutionEnvironment


def pipeline():
    # Create environment
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)

    ds = env.read_text_file('file:///home/user/myfile.json')
    ds.map(lambda i: i).print()

    # Execute job
    env.execute('DynamicStockpilePipeline')


if __name__ == '__main__':
    pipeline()

Answer 2

我通过将 pyenv 3.8 分配为虚拟环境而不是在 PyCharm 中从系统解释器选项分配 pyenv 3.8 来摆脱这个错误。我猜关于 pythonpath 的一些东西被前一个选项破坏了

不能运行基本的 PyFlink 例子

Can't run basic PyFlink example

apache-flink

pyflink

不能 运行 基本的 PyFlink 例子

Can't run basic PyFlink example

apache-flink

pyflink

不能运行基本的 PyFlink 例子