You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "bithw1 (via GitHub)" <gi...@apache.org> on 2024/09/12 08:24:23 UTC
[I] [SUPPORT]Read Kafka and write to Hudi doesn't work [hudi]
bithw1 opened a new issue, #11932:
URL: https://github.com/apache/hudi/issues/11932
Hi,
I have following code snippet, that I want to
1) read the kafka source
2) write the data into Hudi table
3) query the hudi table and print on the console
I am sure that the kafka source can read data from kakfa(I have tested this in the code), But, no data is written into the hudi table directory/file, and also there is only an `Empty set` is print on the console.
Could someone help me out？Thanks very much
```
package org.example.streaming
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, _}
import org.apache.flink.types.Row
object Test002_ReadKafkaAndWriteHudi {
val hudi_table_name = "Test002_ReadKafkaAndWriteHudi"
def main(args: Array[String]): Unit = {
val settings = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build()
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tenv = StreamTableEnvironment.create(env, settings)
val topic = "topic1"
val groupId = "Test001_ReadKafka"
val startupMode = "latest-offset"
val source_ddl =
s"""
CREATE TABLE kafka_table (
a STRING,
b STRING
) WITH (
'connector' = 'kafka',
'topic' = '$topic',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = '$groupId',
'scan.startup.mode' = '$startupMode',
'value.format' = 'json',
'value.json.fail-on-missing-field' = 'false',
'value.fields-include' = 'EXCEPT_KEY'
)
""".stripMargin(' ')
tenv.executeSql(source_ddl)
val target_dll =
s"""
create table ${hudi_table_name} (
a varchar(20),
b varchar(20),
primary key(a) not enforced
)
with (
'connector' = 'hudi'
,'path' = 'file:///D:/data/hudi_demo/${hudi_table_name}'
,'table.type' = 'COPY_ON_WRITE'
,'table.name' = 'test_table'
,'write.insert.drop.duplicates' = 'true'
,'write.recordkey.field' = 'a'
,'write.precombine.field' = 'b'
,'write.shuffle.parallelism' = '1'
,'write.commit.max.retries' = '5'
,'write.cleaner.policy' = 'KEEP_LATEST_COMMITS'
)
""".stripMargin(' ')
tenv.executeSql(target_dll)
val sql =
s"""
insert into $hudi_table_name select a, b from kafka_table
""".stripMargin(' ')
tenv.executeSql(sql)
val query = s"select * from $hudi_table_name"
val result = tenv.sqlQuery(query)
result.execute().print()
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT]Read Kafka and write to Hudi doesn't work [hudi]
Posted by "bithw1 (via GitHub)" <gi...@apache.org>.
bithw1 commented on issue #11932:
URL: https://github.com/apache/hudi/issues/11932#issuecomment-2347772625
Thanks @migeruj ,I will try.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT]Read Kafka and write to Hudi doesn't work [hudi]
Posted by "migeruj (via GitHub)" <gi...@apache.org>.
migeruj commented on issue #11932:
URL: https://github.com/apache/hudi/issues/11932#issuecomment-2347352893
Hi!
Based on your path destination I guess you are on a Windows system.
I tried to test it with `'path' = 'file:///D:/data/hudi_demo/${hudi_table_name}'` and it didn't work and I don't see any error either. So maybe Hudi is trying to use a distributed filesystem based on that URI.
I changed to file:/D:/data/hudi_demo/${hudi_table_name} and started to generate the test_table folder with a .hoodie .hoodie.properties inside. Try with that path or this one D:\data\hudi_demo\${hudi_table_name}.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT]Read Kafka and write to Hudi doesn't work [hudi]
Posted by "bithw1 (via GitHub)" <gi...@apache.org>.
bithw1 commented on issue #11932:
URL: https://github.com/apache/hudi/issues/11932#issuecomment-2347906937
I figured out that checkpoint must be enabled for the streaming write to commit to Hudi table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT]Read Kafka and write to Hudi doesn't work [hudi]
Posted by "bithw1 (via GitHub)" <gi...@apache.org>.
bithw1 closed issue #11932: [SUPPORT]Read Kafka and write to Hudi doesn't work
URL: https://github.com/apache/hudi/issues/11932
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org