ImportError: no se puede importar el nombre ‘OldCsv’ de ‘pyflink.table.descriptors’
Recientemente comencé a utilizar Flink para el procesamiento de datos. Cuando intenté ejecutar la API de tablas para contar hashtags importando pyflink, no pude importar OldCsv y FileSystem desde pyflink.table.descriptors.
También descargué Apache Flink utilizando: pip install apache-flink
Librerías importadas:
from pyflink.table import DataTypes, TableEnvironment, EnvironmentSettings
from pyflink.table.descriptors import Schema, OldCsv, FileSystem
from pyflink.table.expressions import lit
Código:
t_env.get_config().get_configuration().set_string(“parallelism.default”, “1”)
t_env.connect(FileSystem().path(input_file)) \
.with_format(OldCsv()
.field(‘word’, DataTypes.STRING())) \
.with_schema(Schema()
.field(‘word’, DataTypes.STRING())) \
.create_temporary_table(‘Source’)
Error:
File “/home/samarth/Data-Engg/Flink/HashtagCounts/hashtag.py”, line 4, in
from pyflink.table.descriptors import Schema, OldCsv, FileSystem
ImportError: no se puede importar el nombre “OldCsv” de “pyflink.table.descriptors” (/home/samarth/.local/lib/python3.8/site-packages/pyflink/table/descriptors.py)
Link del código anterior:
https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/python/tableapi-tutorial/
davy.ai
Parece que el problema aquí está relacionado con la versión de Flink que se está utilizando. Las clases OldCsv y FileSystem no están disponibles en algunas versiones de pyflink, incluyendo la instalada con pip (paquete apache-flink).
Para solucionar este problema, será necesario descargar y utilizar la última versión de pyflink. Puede hacer esto descargando el paquete pyflink del repositorio PyPI utilizando el comando pip.
Puede probar el siguiente comando para instalar la última versión de pyflink:
Este comando deberá instalar la última versión de pyflink, la cual debería incluir las clases OldCsv y FileSystem.
Una vez que haya instalado la última versión de pyflink, podrá importar las clases OldCsv y FileSystem en su código, y el error debería resolverse.