Hello community!
I'm trying to use FlinkML to train a model on data from a PostgreSQL table and
I get an error when I try to view the output table after model
AttributeError: 'DenseVector' object has no attribute 'get_fields_by_names'
My code:
# Create train source table
t_env.execute_sql(
"""
CREATE TABLE train_source (
id TINYINT,
feature1 FLOAT,
feature2 FLOAT,
feature3 FLOAT,
feature4 FLOAT,
feature5 FLOAT,
label TINYINT
) WITH (
'url' = 'jdbc:postgresql://localhost:5432/train',
'table-name' = 'train',
'username' = 'postgres',
'password' = 'postgres',
'connector' = 'jdbc'
)
"""
)
input_table = t_env.from_path("train_source")
class ListDenseVector(ScalarFunction):
def eval(self, *args):
return Vectors.dense(list(args))
# Map DenseVector to features
t_env.create_temporary_function("dense_map", udf(ListDenseVector(),
result_type=DataTypes.ROW()))
train_table = t_env.sql_query("SELECT id, dense_map(feature1, feature2,
feature3, feature4, feature5) as features, CAST(label as DOUBLE) as label FROM
train_source")
# Train model
logistic_regression = LogisticRegression()
model = logistic_regression.fit(train_table)
output = model.transform(train_table)[0]
# Print result
output.execute().print()
I also tried to change the output type in UDF, but I get an error
TypeError: Invalid returnType: returnType should be DataType or str but is
DenseVectorTypeInfo
________________________________
"This message contains confidential information/commercial secret. If you are
not the intended addressee of this message you may not copy, save, print or
forward it to any third party and you are kindly requested to destroy this
message and notify the sender thereof by email.
Данное сообщение содержит конфиденциальную информацию/информацию, являющуюся
коммерческой тайной. Если Вы не являетесь надлежащим адресатом данного
сообщения, Вы не вправе копировать, сохранять, печатать или пересылать его
каким либо иным лицам. Просьба уничтожить данное сообщение и уведомить об этом
отправителя электронным письмом."
