Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
d682f40
commit b248b08
Showing
15 changed files
with
792 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
69 changes: 69 additions & 0 deletions
69
symmetric-client/src/main/java/org/jumpmind/symmetric/ext/BigQueryDataLoaderFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| package org.jumpmind.symmetric.ext; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| import org.jumpmind.db.platform.DatabaseNamesConstants; | ||
| import org.jumpmind.db.platform.IDatabasePlatform; | ||
| import org.jumpmind.db.platform.bigquery.BigQueryPlatform; | ||
| import org.jumpmind.symmetric.ISymmetricEngine; | ||
| import org.jumpmind.symmetric.db.ISymmetricDialect; | ||
| import org.jumpmind.symmetric.io.BigQueryBulkDatabaseWriter; | ||
| import org.jumpmind.symmetric.io.data.IDataWriter; | ||
| import org.jumpmind.symmetric.io.data.writer.Conflict; | ||
| import org.jumpmind.symmetric.io.data.writer.IDatabaseWriterErrorHandler; | ||
| import org.jumpmind.symmetric.io.data.writer.IDatabaseWriterFilter; | ||
| import org.jumpmind.symmetric.io.data.writer.ResolvedData; | ||
| import org.jumpmind.symmetric.io.data.writer.TransformWriter; | ||
| import org.jumpmind.symmetric.io.stage.IStagingManager; | ||
| import org.jumpmind.symmetric.load.AbstractDataLoaderFactory; | ||
| import org.jumpmind.symmetric.load.IDataLoaderFactory; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.google.cloud.bigquery.BigQuery; | ||
|
|
||
| public class BigQueryDataLoaderFactory extends AbstractDataLoaderFactory implements IDataLoaderFactory { | ||
| protected final Logger log = LoggerFactory.getLogger(getClass()); | ||
|
|
||
| private IStagingManager stagingManager; | ||
|
|
||
| private BigQuery bigquery; | ||
|
|
||
| public BigQueryDataLoaderFactory(ISymmetricEngine engine) { | ||
| this.stagingManager = engine.getStagingManager(); | ||
| this.parameterService = engine.getParameterService(); | ||
| this.bigquery = ((BigQueryPlatform) engine.getSymmetricDialect().getTargetPlatform()).getBigQuery(); | ||
|
|
||
| } | ||
|
|
||
| public String getTypeName() { | ||
| return "bigquery_bulk"; | ||
| } | ||
|
|
||
| public IDataWriter getDataWriter(String sourceNodeId, ISymmetricDialect symmetricDialect, | ||
| TransformWriter transformWriter, | ||
| List<IDatabaseWriterFilter> filters, List<IDatabaseWriterErrorHandler> errorHandlers, | ||
| List<? extends Conflict> conflictSettings, List<ResolvedData> resolvedData) { | ||
|
|
||
| try { | ||
| return new BigQueryBulkDatabaseWriter(symmetricDialect.getPlatform(), symmetricDialect.getTargetPlatform(), | ||
| symmetricDialect.getTablePrefix(), stagingManager, filters, errorHandlers, parameterService, | ||
| buildParameterDatabaseWritterSettings(), this.bigquery); | ||
|
|
||
| } catch (Exception e) { | ||
| log.warn( | ||
| "Failed to create the big query database writer.", | ||
| e); | ||
| if (e instanceof RuntimeException) { | ||
| throw (RuntimeException) e; | ||
| } else { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public boolean isPlatformSupported(IDatabasePlatform platform) { | ||
| return (DatabaseNamesConstants.BIGQUERY.equals(platform.getName())); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
84 changes: 84 additions & 0 deletions
84
symmetric-client/src/main/java/org/jumpmind/symmetric/io/BigQueryBulkDatabaseWriter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| package org.jumpmind.symmetric.io; | ||
|
|
||
|
|
||
| import java.io.File; | ||
| import java.io.FileNotFoundException; | ||
| import java.io.IOException; | ||
| import java.io.OutputStream; | ||
| import java.nio.channels.Channels; | ||
| import java.sql.SQLException; | ||
| import java.util.List; | ||
|
|
||
| import org.jumpmind.db.model.Table; | ||
| import org.jumpmind.db.platform.IDatabasePlatform; | ||
| import org.jumpmind.symmetric.io.data.writer.DatabaseWriterSettings; | ||
| import org.jumpmind.symmetric.io.data.writer.IDatabaseWriterErrorHandler; | ||
| import org.jumpmind.symmetric.io.data.writer.IDatabaseWriterFilter; | ||
| import org.jumpmind.symmetric.io.stage.IStagingManager; | ||
| import org.jumpmind.symmetric.service.IParameterService; | ||
|
|
||
| import com.google.cloud.bigquery.BigQuery; | ||
| import com.google.cloud.bigquery.FormatOptions; | ||
| import com.google.cloud.bigquery.Job; | ||
| import com.google.cloud.bigquery.JobId; | ||
| import com.google.cloud.bigquery.JobStatistics.LoadStatistics; | ||
| import com.google.cloud.bigquery.TableDataWriteChannel; | ||
| import com.google.cloud.bigquery.TableId; | ||
| import com.google.cloud.bigquery.WriteChannelConfiguration; | ||
| import com.google.common.io.Files; | ||
|
|
||
|
|
||
| public class BigQueryBulkDatabaseWriter extends CloudBulkDatabaseWriter { | ||
|
|
||
| BigQuery bigquery; | ||
|
|
||
| public BigQueryBulkDatabaseWriter(IDatabasePlatform symmetricPlatform, IDatabasePlatform targetPlatform, | ||
| String tablePrefix, IStagingManager stagingManager, List<IDatabaseWriterFilter> filters, | ||
| List<IDatabaseWriterErrorHandler> errorHandlers, IParameterService parameterService, DatabaseWriterSettings writerSettings, | ||
| BigQuery bq) throws FileNotFoundException, IOException { | ||
| super(symmetricPlatform, targetPlatform, tablePrefix, stagingManager, filters, errorHandlers, parameterService, writerSettings); | ||
| fieldTerminator = ","; | ||
| bigquery = bq; | ||
| } | ||
|
|
||
| @Override | ||
| public void loadToCloudDatabase() throws SQLException { | ||
| try { | ||
| File csvPath = this.stagedInputFile.getFile(); | ||
|
|
||
| TableId tableId = TableId.of(this.targetTable.getSchema(), this.targetTable.getName()); | ||
| WriteChannelConfiguration writeChannelConfiguration = | ||
| WriteChannelConfiguration.newBuilder(tableId).setFormatOptions(FormatOptions.csv()).setAutodetect(false).setDestinationTable(tableId).build(); | ||
| // The location must be specified; other fields can be auto-detected. | ||
| JobId jobId = JobId.newBuilder() | ||
| .setLocation(bigquery.getOptions().getLocation()) | ||
| .setProject(bigquery.getOptions().getProjectId()).build(); | ||
|
|
||
| TableDataWriteChannel writer = bigquery.writer(jobId, writeChannelConfiguration); | ||
| // Write data to writer | ||
| OutputStream stream = Channels.newOutputStream(writer); | ||
| Files.copy(csvPath, stream); | ||
| stream.close(); | ||
|
|
||
| // Get load job | ||
| Job job = writer.getJob(); | ||
| job = job.waitFor(); | ||
| LoadStatistics stats = job.getStatistics(); | ||
| } catch (Exception ex) { | ||
| throw getPlatform().getSqlTemplate().translate(ex); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| protected Table lookupTableAtTarget(Table sourceTable) { | ||
| return sourceTable; | ||
| } | ||
|
|
||
| @Override | ||
| public void copyToCloudStorage() throws SQLException { | ||
| } | ||
|
|
||
| @Override | ||
| public void cleanUpCloudStorage() throws SQLException { | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.