Apache Beam 实战指南 | 玩转大数据存储 HDFSIO

HadoopFileSystem(Configuration configuration) throws IOException { this.fileSystem = org.apache.hadoop.fs.FileSystem.newInstance(configuration);
} 

@Override
public int read(ByteBuffer dst) throws IOException {
if (closed) {
throw new IOException("Channel is closed");
}
// O length read must be supported
int read = 0;
// We avoid using the ByteBuffer based read for Hadoop because some FSDataInputStream
// implementations are not ByteBufferReadable,
// See https://issues.apache.org/jira/browse/HADOOP-14603
if (dst.hasArray()) {// does the same as inputStream.read(dst):// stores up to dst.remaining() bytes into dst.array() starting at dst.position().// But dst can have an offset with its backing array hence the + dst.arrayOffset()
read = inputStream.read(dst.array(), dst.position() + dst.arrayOffset(), dst.remaining());} else {// TODO: Add support for off heap ByteBuffers in case the underlying FSDataInputStream// does not support reading from a ByteBuffer.read = inputStream.read(dst);}if (read > 0) {dst.position(dst.position() + read);}
return read;
}

pipeline.apply(JdbcIO.<KV<Integer, String>>read()
//数据库JDBCIO的配置
.withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
//mysql的驱动
"com.mysql.jdbc.Driver", "jdbc:mysql://hostname:3306/mydb")
//MySQL的用户名
.withUsername("username")
//MySQL的密码
.withPassword("password"))
//查询SQL语句
.withQuery("select id,name from Person")
//返回数据的编码
.withCoder(KvCoder.of(BigEndianIntegerCoder.of(), StringUtf8Coder.of()))
//映射返回数据
.withRowMapper(new JdbcIO.RowMapper<KV<Integer, String>>() {
public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
return KV.of(resultSet.getInt(1), resultSet.getString(2));
}
})
);

//添加带条件的SQL读取.withQuery("select id,name from Person where name = ?")//数据编码.withCoder(KvCoder.of(BigEndianIntegerCoder.of(), StringUtf8Coder.of()))//实例化JdbcIO参数类.withStatementPreparator(new JdbcIO.StatementPreparator() {//传入执行Statementpublic void setParameters(PreparedStatement preparedStatement)throws Exception {
//传递条件参数preparedStatement.setString(1, "Darwin");}}).withRowMapper(new JdbcIO.RowMapper<KV<Integer, String>>() {

/**
* Provide a maximum size in number of SQL statenebt for the batch. Default is 1000.
** @param batchSize maximum batch size in number of statements*/public Write<T> withBatchSize(long batchSize) {checkArgument(batchSize > 0, "batchSize must be > 0, but was %d", batchSize);return toBuilder().setBatchSize(batchSize).build();}

private void processRecord(T record, PreparedStatement preparedStatement) {
try {
preparedStatement.clearParameters();spec.getPreparedStatementSetter().setParameters(record, preparedStatement);preparedStatement.addBatch();} catch (Exception e) {
throw new RuntimeException(e);}}

CREATE TABLE `test` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`keystr` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
`count` int(11) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-core-java</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-direct-java</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-jdbc</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-hadoop-file-system</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-hdfs</artifactId>
<version>0.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-hadoop-common</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>jdk.tools</groupId>
<artifactId>jdk.tools</artifactId>
<version>1.8</version>
<scope>system</scope>
<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.7.2</version>
<scope>provided</scope>
</dependency>
<!-- 不加上会报 版本比这是由于Jackson这个工具库的版本不一致导致的
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Incompatible Jackson version:-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.2.0</version>
<exclusions>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.2.0</version>
<exclusions>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.6.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.6.7</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>2.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-jdbc</artifactId>
<version>2.4.0</version>
</dependency>
<!-- 不加上会出现找不到驱动问题 -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.17</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-spark</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.2</version>
</dependency> 

public static void main(String[] args) {
//配置Hdfs配置
Configuration conf = new Configuration();
//配置Hdfs 的地址
conf.set("fs.default.name", "hdfs://192.168.220.140:9000");
//设置管道
HadoopFileSystemOptions options = PipelineOptionsFactory.fromArgs(a
 
rgs).withValidation().as(HadoopFileSystemOptions.class);
 
options.setHdfsConfiguration(ImmutableList.of(conf));
//这里可以指定 任意平台，我这里本地测试用的本地Runner
options.setRunner(DirectRunner.class);
// options.setRunner(SparkRunner.class);
// options.setRunner(FlinkRunner.class);
Pipeline pipeline = Pipeline.create(options);
// 读取Hdfs的数据跟读取TextIO一样的。
 
PCollection<String> resq = pipeline.apply(TextIO.read().from("hdf
 
s://192.168.220.140:9000/user/lenovo/testfile/test.txt")).apply
 
("ExtractWords",ParDo.of(new DoFn<String, String>()
{
private static final long serialVersionUID = 1L;
 
@ProcessElement
public void processElement(ProcessContext c) {
// 根据空格进行读取数据，里面可以用Luma 表达式写
for (String word : c.element().split(" ")) {
if (!word.isEmpty()) {
c.output(word);
System.out.println(word + "\n");
}
}
}
}));
 
PCollection<String> windowedEvents = resq.apply(Window.<String>
 
into(FixedWindows.of(Duration.standardSeconds(5))));
//把数据出现频率做个统计
PCollection<KV<String, Long>> wordcount = windowedEvents.apply(Count.<String>perElement());
//写入mysql,调用 Beam 的JdbcIO 进行写入
wordcount.apply(JdbcIO.<KV<String, Long>>write()
 
.withDataSourceConfiguration(JdbcIO.DataSourceConfiguration
//mysql 的驱动和连接
.create("com.mysql.jdbc.Driver", "jdbc:mysql://192.168.220.
 
140:3306/db1?useUnicode=true&characterEncoding=UTF-8&allowM ultiQueries=true")
//mysql 的用户名和密码
.withUsername("root").withPassword("123456"))
//SQL语句
.withStatement("insert into test (keystr,count) values(?,?)")
.withPreparedStatementSetter(new JdbcIO.PreparedStatementSe
 
tter<KV<String, Long>>() {
 
private static final long serialVersionUID = 1L;
public void setParameters(KV<String, Long> element, PreparedSta
 
tement query) throws SQLException {
//传递的参数
query.setString(1, element.getKey());
query.setLong(2, element.getValue());
}
}));
pipeline.run().waitUntilFinish();
System.exit(0);
}

./spark-submit --master spark://master:7077 --class com.BeamHdfsToMySQL.HdfsToMySQL.HTM /home/beamsparkHdfstoMysql.jar