无法运行基本的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

我通过分配pyenv3.8作为虚拟环境，而不是在PyCharm中分配pyenv3.8from系统解释器选项，从而消除了这个错误。我猜上一个选项破坏了pythonpath中的某些内容。

Answer 2

也许您可以在报告错误时发布代码。

根据上下文，这里报告的错误可能是使用错误，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()