Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[HUDI-7645] Optimize BQ sync tool for MDT #11065

Merged
merged 1 commit into from
Apr 25, 2024
Merged

[HUDI-7645] Optimize BQ sync tool for MDT #11065

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/BigQuerySyncTool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@
public class BigQuerySyncTool extends HoodieSyncTool {

private static final Logger LOG = LoggerFactory.getLogger(BigQuerySyncTool.class);
private static final String SUFFIX_MANIFEST = "_manifest";
private static final String SUFFIX_VERSIONS = "_versions";

private final BigQuerySyncConfig config;
private final String tableName;
Expand All @@ -69,8 +71,8 @@ public BigQuerySyncTool(Properties props) {
super(props);
this.config = new BigQuerySyncConfig(props);
this.tableName = config.getString(BIGQUERY_SYNC_TABLE_NAME);
this.manifestTableName = tableName + "_manifest";
this.versionsTableName = tableName + "_versions";
this.manifestTableName = tableName + SUFFIX_MANIFEST;
this.versionsTableName = tableName + SUFFIX_VERSIONS;
this.snapshotViewName = tableName;
this.bqSyncClient = new HoodieBigQuerySyncClient(config);
// reuse existing meta client if not provided (only test cases will provide their own meta client)
Expand All @@ -85,8 +87,8 @@ public BigQuerySyncTool(Properties props) {
super(properties);
this.config = new BigQuerySyncConfig(props);
this.tableName = config.getString(BIGQUERY_SYNC_TABLE_NAME);
this.manifestTableName = tableName + "_manifest";
this.versionsTableName = tableName + "_versions";
this.manifestTableName = tableName + SUFFIX_MANIFEST;
this.versionsTableName = tableName + SUFFIX_VERSIONS;
this.snapshotViewName = tableName;
this.bqSyncClient = bigQuerySyncClient;
this.metaClient = metaClient;
Expand Down Expand Up @@ -115,7 +117,7 @@ public void syncHoodieTable() {

private boolean tableExists(HoodieBigQuerySyncClient bqSyncClient, String tableName) {
if (bqSyncClient.tableExists(tableName)) {
LOG.info(tableName + " already exists. Skip table creation.");
LOG.info("{} already exists. Skip table creation.", tableName);
return true;
}
return false;
Expand Down
19 changes: 13 additions & 6 deletions ...c/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/ManifestFileWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ public synchronized void writeManifestFile(boolean useAbsolutePath) {
LOG.warn("No base file to generate manifest file.");
return;
} else {
LOG.info("Writing base file names to manifest file: " + baseFiles.size());
LOG.info("Writing base file names to manifest file: {}", baseFiles.size());
}
final StoragePath manifestFilePath = getManifestFilePath(useAbsolutePath);
try (OutputStream outputStream = metaClient.getStorage().create(manifestFilePath, true);
Expand All @@ -85,16 +85,23 @@ public synchronized void writeManifestFile(boolean useAbsolutePath) {
public static Stream<String> fetchLatestBaseFilesForAllPartitions(HoodieTableMetaClient metaClient,
boolean useFileListingFromMetadata, boolean useAbsolutePath) {
try {
List<String> partitions = FSUtils.getAllPartitionPaths(new HoodieLocalEngineContext(metaClient.getHadoopConf()),
metaClient.getBasePath(), useFileListingFromMetadata);
LOG.info("Retrieve all partitions: " + partitions.size());

Configuration hadoopConf = metaClient.getHadoopConf();
HoodieLocalEngineContext engContext = new HoodieLocalEngineContext(hadoopConf);
HoodieMetadataFileSystemView fsView = new HoodieMetadataFileSystemView(engContext, metaClient,
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(),
HoodieMetadataConfig.newBuilder().enable(useFileListingFromMetadata).build());
return partitions.parallelStream().flatMap(partition -> fsView.getLatestBaseFiles(partition).map(useAbsolutePath ? HoodieBaseFile::getPath : HoodieBaseFile::getFileName));
Stream<HoodieBaseFile> allLatestBaseFiles;
if (useFileListingFromMetadata) {
LOG.info("Fetching all base files from MDT.");
fsView.loadAllPartitions();
allLatestBaseFiles = fsView.getLatestBaseFiles();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like fsView.getLatestBaseFiles() only returns already loaded file-groups in the view so some partitions may not be loaded at all. Can you check if this is your intended behavior ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, you are right, fsView.getLatestBaseFiles() only returns already loaded file-groups in the view. But I don't see any other approach to load all latest files in one call to HoodieMetadataFileSystemView/HoodieTableMetadata. It would be great if you or @nsivabalan (as reporter of this task) give me some advice.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i've found fsView.loadAllPartitions() to load all partitions in one call, and now all file-groups are loaded in the view before getting latest base files by fsView.getLatestBaseFiles().

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Makes sense. LGTM

} else {
List<String> partitions = FSUtils.getAllPartitionPaths(new HoodieLocalEngineContext(metaClient.getHadoopConf()),
metaClient.getBasePathV2().toString(), false);
LOG.info("Retrieve all partitions from fs: {}", partitions.size());
allLatestBaseFiles = partitions.parallelStream().flatMap(fsView::getLatestBaseFiles);
}
return allLatestBaseFiles.map(useAbsolutePath ? HoodieBaseFile::getPath : HoodieBaseFile::getFileName);
} catch (Exception e) {
throw new HoodieException("Error in fetching latest base files.", e);
}
Expand Down
Loading