使用flink迁移数据

在公司中，遇到这样一个业务，需要将数据库A从oracle迁移到pg数据库，原本让实习生去实现了这样的一个工具，但是最后他写出来的工具存在较大问题。

正好最近在学习spark、flink等流式处理框架，那么我们就用flink来处理这样一个需求吧：

1、主类：

package com.ogj.flink;import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;import java.util.concurrent.TimeUnit;/*** @author ouguangji*/
public class DbMove {public static void main(String[] args) throws Exception {ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();DataSource<Row> dataSource = env.createInput(JDBCInputFormat.buildJDBCInputFormat().setDBUrl(DBContent.SourceDB.url).setDrivername(DBContent.PGDRIVER).setUsername(DBContent.SourceDB.username).setPassword(DBContent.SourceDB.password).setQuery("select task_name,file_path from cloud_task").setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO)).finish());dataSource.output(JDBCOutputFormat.buildJDBCOutputFormat().setDBUrl(DBContent.DstDB.url).setDrivername(DBContent.PGDRIVER).setUsername(DBContent.DstDB.username).setPassword(DBContent.DstDB.password).setQuery("insert into test(task_name,file_path) values(?,?)").finish());env.execute("db move");System.out.println("写入数据中");TimeUnit.SECONDS.sleep(5);//查询出来DataSource<Row> read = env.createInput(JDBCInputFormat.buildJDBCInputFormat().setDBUrl(DBContent.DstDB.url).setDrivername(DBContent.PGDRIVER).setUsername(DBContent.DstDB.username).setPassword(DBContent.DstDB.password).setQuery("select task_name,file_path from test").setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO)).finish());System.out.println("read dst dataSource");read.print();System.out.println("=========end==========");}
}

2、数据库配置类

package com.ogj.flink;public class DBContent {public static final String MYSQLDRIVER = "com.mysql.jdbc.Driver";public static final String PGDRIVER = "org.postgresql.Driver";public static class SourceDB {public static String url = "jdbc:postgresql://127.0.0.1:5432/dmt_url?currentSchema=schema_name";public static String username = "postgres";public static String password = "123456";}public static class DstDB {public static String url = "jdbc:postgresql://127.0.0.1:5432/RequestMonitor";public static String username = "postgres";public static String password = "123456";}
}

3、maven打包配置：

<build>
<!--        &lt;!&ndash;scala待编译的文件目录&ndash;&gt;-->
<!--        <sourceDirectory>src/main/scala</sourceDirectory>-->
<!--        <testSourceDirectory>src/test/scala</testSourceDirectory>-->
<!--        &lt;!&ndash;scala插件&ndash;&gt;--><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.2.2</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals></execution></executions></plugin><!--            <plugin>-->
<!--                <groupId>net.alchim31.maven</groupId>-->
<!--                <artifactId>scala-maven-plugin</artifactId>-->
<!--                <version>3.2.2</version>-->
<!--                <executions>-->
<!--                    <execution>-->
<!--                        <goals>-->
<!--                            <goal>compile</goal>-->
<!--                            <goal>testCompile</goal>-->
<!--                        </goals>-->
<!--                        <configuration>-->
<!--                            <args>-->
<!--                                &lt;!&ndash;<arg>-make:transitive</arg>&ndash;&gt;&lt;!&ndash;scala2.11 netbean不支持这个参数&ndash;&gt;-->
<!--                                <arg>-dependencyfile</arg>-->
<!--                                <arg>${project.build.directory}/.scala_dependencies</arg>-->
<!--                            </args>-->
<!--                        </configuration>-->
<!--                    </execution>-->
<!--                </executions>-->
<!--            </plugin>--><!-- 这是个编译java代码的 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.0</version><configuration><source>8</source><target>8</target><encoding>UTF-8</encoding></configuration><executions><execution><phase>compile</phase><goals><goal>compile</goal></goals></execution></executions></plugin><!--            &lt;!&ndash; 这是个编译scala代码的 &ndash;&gt;-->
<!--            <plugin>-->
<!--                <groupId>net.alchim31.maven</groupId>-->
<!--                <artifactId>scala-maven-plugin</artifactId>-->
<!--                <version>3.2.1</version>-->
<!--                <executions>-->
<!--                    <execution>-->
<!--                        <id>scala-compile-first</id>-->
<!--                        <phase>process-resources</phase>-->
<!--                        <goals>-->
<!--                            <goal>add-source</goal>-->
<!--                            <goal>compile</goal>-->
<!--                        </goals>-->
<!--                    </execution>-->
<!--                </executions>-->
<!--            </plugin>--><!--manven打包插件--><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>2.4.3</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><filters><filter><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"><resource>reference.conf</resource></transformer><transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>cn.itcast.rpc.Master</mainClass> <!--main方法--></transformer></transformers></configuration></execution></executions></plugin></plugins></build>

运行完毕就是这样的啦：

使用flink迁移数据相关推荐

分库分表：如何做到永不迁移数据和避免热点？
点击关注公众号,Java干货及时送达中大型项目中,一旦遇到数据量比较大,小伙伴应该都知道就应该对数据进行拆分了.有垂直和水平两种. 垂直拆分比较简单,也就是本来一个数据库,数据量大之后,从业务角度进 ...
你知道怎么分库分表吗？如何做到永不迁移数据和避免热点吗？
点击关注公众号,Java干货及时送达前言中大型项目中,一旦遇到数据量比较大,小伙伴应该都知道就应该对数据进行拆分了.有垂直和水平两种. 垂直拆分比较简单,也就是本来一个数据库,数据量大之后,从业务 ...
迁移数据中心的三大风险与应对策略
数据中心迁移是项复杂工程,可能很难向为迁移活动拨款的企业高管解释清楚,需要了解并管理相关的业务运营风险.下面我们分析一下与管理数据中心有关的几个难题和风险. 服务可用性数据中心的主要目的就是,托管运 ...
迁移数据文件到ASM【转】
1.迁移数据文件到ASM 1)数据库一致性情况下迁移: 将数据库启动到mount状态,生成rman copy 语句,然后在rman中执行: SQL> startup mount SQL& ...
分库分表这样玩，可以永不迁移数据、避免热点
作者 | 老顾聊技术来源 | http://www.toutiao.com/i6677459303055491597 中大型项目中,一旦遇到数据量比较大,小伙伴应该都知道就应该对数据进行拆分了.有垂 ...
分库分表？如何做到永不迁移数据和避免热点？
一.前言中大型项目中,一旦遇到数据量比较大,小伙伴应该都知道就应该对数据进行拆分了.有垂直和水平两种. 垂直拆分比较简单,也就是本来一个数据库,数据量大之后,从业务角度进行拆分多个库.如下图,独立的 ...
rediscluster全局数据_redis迁移，从redis cluster集群迁移数据到另外一个redis cluster集群...
redis迁移,redis集群迁移$ yum install automake libtool autoconf bzip2 -y $ git clone https://github.com/vip ...
flink 写入到es_《从0到1学习Flink》—— Flink 写入数据到 Kafka
前言之前文章 <从0到1学习Flink>-- Flink 写入数据到 ElasticSearch 写了如何将 Kafka 中的数据存储到 ElasticSearch 中,里面其实就已经用 ...
ASP.NET Core Identity 迁移数据 - ASP.NET Core 基础教程 - 简单教程，简单编程
ASP.NET Core Identity 迁移数据 - ASP.NET Core 基础教程 - 简单教程,简单编程原文:ASP.NET Core Identity 迁移数据 - ASP.NET C ...

使用flink迁移数据

1、主类：

2、数据库配置类

3、maven打包配置：

使用flink迁移数据相关推荐

最新文章

热门文章