flink_sql create table and to save hive

[lijihuaxiaomo]

Tips before filing an issue

• Have you gone through our FAQs?

• Join the mailing list to engage in conversations and get faster support at [email protected].

• If you have triaged this as a bug, then file an issue directly.

Describe the problem you faced

A clear and concise description of the problem.

To Reproduce

Steps to reproduce the behavior:

Expected behavior

A clear and concise description of what you expected to happen.

Environment Description

• Hudi version : 0.14

• Spark version :

• Hive version :3.1.3

• Hadoop version :3.4.1

Flink SQL> CREATE TABLE t103(

uuid VARCHAR(20),
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
partition VARCHAR(20)
)
PARTITIONED BY (partition)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://10.10.127.151:9000/bi/hudi/t103',
'table.type' = 'COPY_ON_WRITE',
'hive_sync.enable' = 'true',
'hive_sync.mode' = 'hms',
'hive_sync.metastore.uris' = 'thrift://10.10.127.151:9083'
);
[INFO] Execute statement succeed.

Flink SQL> INSERT INTO t103 VALUES

('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'),
('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),
('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),
('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),
('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),
('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),
('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),
('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.runtime.rest.util.RestClientException: [Internal server error., <Exception on server side:
org.apache.flink.runtime.rpc.exceptions.RecipientUnreachableException: Could not send message [LocalFencedMessage(00000000000000000000000000000000, LocalRpcInvocation(JobMasterGateway.requestJob(Time)))] from sender [Actor[akka://flink/temp/jobmanager_2$mr]] to recipient [Actor[akka://flink/user/rpc/jobmanager_2#-1187081014]], because the recipient is unreachable. This can either mean that the recipient has been terminated or that the remote RpcService is currently not reachable.

Flink SQL>

[danny0405]

@lijihuaxiaomo Hi, did you already solve the issues?

[rangareddy]

Hi @lijihuaxiaomo

To further investigate this issue:

1. Verify the presence of hive-exec.jar: Please confirm that hive-exec.jar is located in the Flink's lib directory (or designated JAR location). If it's missing, download and place it there.
2. Examine Flink logs: Please review the logs for both the Flink JobManager and TaskManagers for any relevant error messages or warnings.
3. Test without Hive synchronization: Temporarily remove all Hive synchronization parameters from your Flink job configuration and re-run the job. This will help determine if the issue is related to Hive synchronization. Observe whether the job now functions as expected.
4. Check for network connectivity problems: As a final step, please investigate any potential network connectivity issues that might be affecting communication between Flink components or external systems.

[rangareddy]

Hi @danny0405

I was able to replicate this issue using Hudi 1.0.0 and Flink 1.17.2. I am attaching the Sample code and Flink logs here.

bash $FLINK_HOME/bin/sql-client.sh embedded -j /opt/flink/lib/hudi-flink1.17-bundle-1.0.0-rc1.jar shell

set sql-client.execution.result-mode = tableau;

CREATE CATALOG hudi_hive_catalog WITH (
  'type' = 'hudi',
  'mode' = 'hms',
  'table.external' = 'true',
  'default-database' = 'default',
  'hive.conf.dir' = '/opt/flink/conf',
  'catalog.path' = 's3a://warehouse/hudi_hive_catalog'
);

USE CATALOG hudi_hive_catalog;

CREATE DATABASE IF NOT EXISTS hudi_db;

USE hudi_db;

CREATE TABLE IF NOT EXISTS hudi_table(
    uuid VARCHAR(20),
    name VARCHAR(10),
    age INT,
    ts TIMESTAMP(3),
    `partition` VARCHAR(20)
)
PARTITIONED BY (`partition`)
WITH (
  'connector' = 'hudi',
  'path' = 's3a://warehouse/hudi_db/hudi_table',
  'table.type' = 'COPY_ON_WRITE'
);

INSERT INTO hudi_table VALUES
    ('id1','Alex',23,TIMESTAMP '1970-01-01 00:00:01','par1'),
    ('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),
    ('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),
    ('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),
    ('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),
    ('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),
    ('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),
    ('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');

flink--sql-client-28d1fd5c5bee.log
flink--standalonesession-1-28d1fd5c5bee.log
flink--taskexecutor-0-e3123ef582b2 (1).log

[rangareddy]

Hi @danny0405

Are you able to check the attached logs to find out the cause?