You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@seatunnel.apache.org by GitBox <gi...@apache.org> on 2022/10/24 06:30:37 UTC
[GitHub] [incubator-seatunnel] Carl-Zhou-CN opened a new pull request, #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Carl-Zhou-CN opened a new pull request, #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169
…oad file is loaded
<!--
Thank you for contributing to SeaTunnel! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
## Contribution Checklist
- Make sure that the pull request corresponds to a [GITHUB issue](https://github.com/apache/incubator-seatunnel/issues).
- Name the pull request in the form "[Feature] [component] Title of the pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
- Minor fixes should be named following this pattern: `[hotfix] [docs] Fix typo in README.md doc`.
-->
## Purpose of this pull request
<!-- Describe the purpose of this pull request. For example: This pull request adds checkstyle plugin.-->
## Check list
* [ ] Code changed are covered with tests, or it does not need tests for reason:
* [ ] If any new Jar binary package adding in your PR, please add License Notice according
[New License Guide](https://github.com/apache/incubator-seatunnel/blob/dev/docs/en/contribution/new-license.md)
* [ ] If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs
close #3168
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer merged pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
TyrantLucifer merged PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Hisoka-X commented on pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
Hisoka-X commented on PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169#issuecomment-1302866709
Hi merge from dev again, the ci will pass. Thanks!
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Carl-Zhou-CN commented on a diff in pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
Carl-Zhou-CN commented on code in PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169#discussion_r1006364175
##########
seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hbase/src/main/scala/org/apache/seatunnel/spark/hbase/sink/Hbase.scala:
##########
@@ -132,31 +133,27 @@ class Hbase extends SparkBatchSink with Logging {
},
stagingDir)
- val load = new LoadIncrementalHFiles(hbaseConf)
- val table = hbaseConn.getTable(tableName)
- load.doBulkLoad(
- new Path(stagingDir),
- hbaseConn.getAdmin,
- table,
- hbaseConn.getRegionLocator(tableName))
+ if (fs.exists(stagingPath)) {
+ val load = new LoadIncrementalHFiles(hbaseConf)
+ val table = hbaseConn.getTable(tableName)
+ load.doBulkLoad(
+ stagingPath,
+ hbaseConn.getAdmin,
+ table,
+ hbaseConn.getRegionLocator(tableName))
+ }
} finally {
if (hbaseConn != null) {
hbaseConn.close()
}
-
- cleanUpStagingDir(stagingDir)
+ cleanUpStagingDir(stagingPath, fs)
}
}
- private def cleanUpStagingDir(stagingDir: String): Unit = {
- val stagingPath = new Path(stagingDir)
- val fs = stagingPath.getFileSystem(hbaseContext.config)
+ private def cleanUpStagingDir(stagingPath: Path, fs: FileSystem): Unit = {
if (!fs.delete(stagingPath, true)) {
- logWarning(s"clean staging dir $stagingDir failed")
- }
- if (fs != null) {
- fs.close()
Review Comment:
Because it cannot be closed at this time, closing will cause the fs object held by Spark to be closed, resulting in an abnormal exit of SparkContext.stop()
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Carl-Zhou-CN commented on pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
Carl-Zhou-CN commented on PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169#issuecomment-1302870118
> Hi merge from dev again, the ci will pass. Thanks!
ok
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169#discussion_r1006365269
##########
seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hbase/src/main/scala/org/apache/seatunnel/spark/hbase/sink/Hbase.scala:
##########
@@ -132,31 +133,27 @@ class Hbase extends SparkBatchSink with Logging {
},
stagingDir)
- val load = new LoadIncrementalHFiles(hbaseConf)
- val table = hbaseConn.getTable(tableName)
- load.doBulkLoad(
- new Path(stagingDir),
- hbaseConn.getAdmin,
- table,
- hbaseConn.getRegionLocator(tableName))
+ if (fs.exists(stagingPath)) {
+ val load = new LoadIncrementalHFiles(hbaseConf)
+ val table = hbaseConn.getTable(tableName)
+ load.doBulkLoad(
+ stagingPath,
+ hbaseConn.getAdmin,
+ table,
+ hbaseConn.getRegionLocator(tableName))
+ }
} finally {
if (hbaseConn != null) {
hbaseConn.close()
}
-
- cleanUpStagingDir(stagingDir)
+ cleanUpStagingDir(stagingPath, fs)
}
}
- private def cleanUpStagingDir(stagingDir: String): Unit = {
- val stagingPath = new Path(stagingDir)
- val fs = stagingPath.getFileSystem(hbaseContext.config)
+ private def cleanUpStagingDir(stagingPath: Path, fs: FileSystem): Unit = {
if (!fs.delete(stagingPath, true)) {
- logWarning(s"clean staging dir $stagingDir failed")
- }
- if (fs != null) {
- fs.close()
Review Comment:
Got it, thank you for your contribution.
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Carl-Zhou-CN commented on a diff in pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
Carl-Zhou-CN commented on code in PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169#discussion_r1006364175
##########
seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hbase/src/main/scala/org/apache/seatunnel/spark/hbase/sink/Hbase.scala:
##########
@@ -132,31 +133,27 @@ class Hbase extends SparkBatchSink with Logging {
},
stagingDir)
- val load = new LoadIncrementalHFiles(hbaseConf)
- val table = hbaseConn.getTable(tableName)
- load.doBulkLoad(
- new Path(stagingDir),
- hbaseConn.getAdmin,
- table,
- hbaseConn.getRegionLocator(tableName))
+ if (fs.exists(stagingPath)) {
+ val load = new LoadIncrementalHFiles(hbaseConf)
+ val table = hbaseConn.getTable(tableName)
+ load.doBulkLoad(
+ stagingPath,
+ hbaseConn.getAdmin,
+ table,
+ hbaseConn.getRegionLocator(tableName))
+ }
} finally {
if (hbaseConn != null) {
hbaseConn.close()
}
-
- cleanUpStagingDir(stagingDir)
+ cleanUpStagingDir(stagingPath, fs)
}
}
- private def cleanUpStagingDir(stagingDir: String): Unit = {
- val stagingPath = new Path(stagingDir)
- val fs = stagingPath.getFileSystem(hbaseContext.config)
+ private def cleanUpStagingDir(stagingPath: Path, fs: FileSystem): Unit = {
if (!fs.delete(stagingPath, true)) {
- logWarning(s"clean staging dir $stagingDir failed")
- }
- if (fs != null) {
- fs.close()
Review Comment:
Because it cannot be closed at this time, closing will cause the fs object held by Spark to be closed, resulting in an abnormal exit of SparkContext
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169#issuecomment-1292917155
Please rebase dev to your branch and restart CI.
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #3169:
URL: https://github.com/apache/incubator-seatunnel/pull/3169#discussion_r1006361766
##########
seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hbase/src/main/scala/org/apache/seatunnel/spark/hbase/sink/Hbase.scala:
##########
@@ -132,31 +133,27 @@ class Hbase extends SparkBatchSink with Logging {
},
stagingDir)
- val load = new LoadIncrementalHFiles(hbaseConf)
- val table = hbaseConn.getTable(tableName)
- load.doBulkLoad(
- new Path(stagingDir),
- hbaseConn.getAdmin,
- table,
- hbaseConn.getRegionLocator(tableName))
+ if (fs.exists(stagingPath)) {
+ val load = new LoadIncrementalHFiles(hbaseConf)
+ val table = hbaseConn.getTable(tableName)
+ load.doBulkLoad(
+ stagingPath,
+ hbaseConn.getAdmin,
+ table,
+ hbaseConn.getRegionLocator(tableName))
+ }
} finally {
if (hbaseConn != null) {
hbaseConn.close()
}
-
- cleanUpStagingDir(stagingDir)
+ cleanUpStagingDir(stagingPath, fs)
}
}
- private def cleanUpStagingDir(stagingDir: String): Unit = {
- val stagingPath = new Path(stagingDir)
- val fs = stagingPath.getFileSystem(hbaseContext.config)
+ private def cleanUpStagingDir(stagingPath: Path, fs: FileSystem): Unit = {
if (!fs.delete(stagingPath, true)) {
- logWarning(s"clean staging dir $stagingDir failed")
- }
- if (fs != null) {
- fs.close()
Review Comment:
Why remove this?
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Carl-Zhou-CN closed pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
Posted by GitBox <gi...@apache.org>.
Carl-Zhou-CN closed pull request #3169: [Bug][Connector-V1-Spark-Hbase] When the written df is empty, the directory does not exist when the load file is loaded
URL: https://github.com/apache/incubator-seatunnel/pull/3169
--
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@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org