HDFS to mongo, convert HDFS ID to _id and write it to mongo. Field type cannot be converted to mongo's object_id
- Dominant language
- Java
- Stars
- 9.7k
- Forks
- 2.4k
- Avg merge
- 3d 9h
- Merged PRs (30d)
- 204
Description
### Search before asking
- [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues.
### What happened
hdfs到mongo,把hdfs的id转为_id写入mongo,字段类型不能转为mongo的object_id
hdfs的id是string,但是符合mongo_id的格式 62ea95971961387b795a423b.
CREATE TABLE test.cts_test2 (
id STRING COMMENT '主键',
ycg_en STRING COMMENT 'ycg_en',
`date` BIGINT COMMENT '时间戳(毫秒)',
date_tm TIMESTAMP,
date_da STRING COMMENT '日期(yyyy-MM-dd)',
zj DOUBLE
)
STORED AS PARQUET;
insert overwrite table test.cts_test2
select "689ae9220cb28d1bcd2ce5b8",1,1610323200000,1610323200000,"2025",2.2
union all
select "62ea95971961387b795a423b",2,1610323200000,1610323200000,"2025",2.3;
### SeaTunnel Version
2.3.11版本
### SeaTunnel Config
```conf
env {
parallelism = 1
job.mode = "BATCH"
}
source {
HdfsFile {
path = "/user/hive/warehouse/test.db/cts_test2"
file_format_type = "parquet"
fs.defaultFS = "hdfs://m1"
plugin_output = "Hdfs_source1"
}
}
transform {
FieldRename {
plugin_input = "Hdfs_source1"
plugin_output = "tran1"
replacements_with_regex = [
{
replace_from = "id"
replace_to = "_id"
}
]
}
FieldMapper {
plugin_input = "tran1"
plugin_output = "tran2"
field_mapper = {
_id = _id
ycg_en=ycg_en
date=date
date_tm=date_tm
date_da=date_da
zj=zj
}
}
}
sink {
Mongodb {
plugin_input = "tran2"
uri =
database = "test"
collection =
upsert-enable=true
primary-key = ["_id"]
buffer-flush.max-rows=10000
}
}
```
### Running Command
```shell
sh bin/seatunnel.sh --config ./conf/hdfs_2_mongo3.conf --async -n hdfs_2_mongo3
```
### Error Exception
```log
这个是吐到mongo的数据:
[{'_id': '689ae9220cb28d1bcd2ce5b8', 'date': 1610323200000, 'date_da': '2025', 'date_tm': datetime.datetime(2021, 1, 11, 0, 0), 'ycg_en': '1', 'zj': 2.2}, {'_id': '62ea95971961387b795a423b', 'date': 1610323200000, 'date_da': '2025', 'date_tm': datetime.datetime(2021, 1, 11, 0, 0), 'ycg_en': '2', 'zj': 2.3}]
```
### Zeta or Flink or Spark Version
2.3.11版本
### Java or Scala Version
java version "1.8.0_181"
### Screenshots
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
No contributing guide indexed for this repository
Research direction
Reproduce the supplied HdfsFile-to-Mongodb batch configuration on SeaTunnel 2.3.11, then inspect the MongoDB sink path that handles the primary-key `_id` value. Compare the two sample IDs and verify that a string matching MongoDB's ObjectId format is written with the intended MongoDB type while other values retain correct behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- hadoop, java, mongodb
- Domain
- data, database
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100