- 浏览: 83048 次
- 性别:
- 来自: 郑州
文章分类
- 全部博客 (69)
- java (28)
- linux (6)
- redis (4)
- C# (3)
- 架构 (10)
- java ee (1)
- web (1)
- 操作系统 (7)
- sqlserver (1)
- android (2)
- Hadoop (12)
- 大数据 (21)
- 分布式 事务 消息 (10)
- linux mysql (1)
- 数据库 (3)
- 关于hadoop之bootshell使用 (2)
- 关于hbase---HTableInterfaceFactory (1)
- Spring (3)
- Hbase (5)
- jstorm (10)
- nginx (1)
- 分布式 (1)
- 区块链 (3)
- dubbo (1)
- nacos (1)
- 阿里 (1)
- go (3)
- 缓存 (1)
- memcached (1)
- ssdb (1)
- 源码 (1)
最新评论
-
想个可以用的名字:
楼主,能不能给发一份源代码,1300246542@qqq.co ...
spring+websocket的使用 -
wahahachuang5:
web实时推送技术使用越来越广泛,但是自己开发又太麻烦了,我觉 ...
websocket -
dalan_123:
前提是你用的是spring mvc 才需要加的1、在web.x ...
spring+websocket的使用 -
string2020:
CharacterEncodingFilter这个filter ...
spring+websocket的使用
一、测试
public class MrBatchApp {
// Log
private static final Log log = LogFactory.getLog(MrBatchApp.class);
//
public static void main(String[] args) throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
System.out.println("TEST");
// 加载对应的xml配置文件
AbstractApplicationContext context = new ClassPathXmlApplicationContext("classpath:/META-INF/spring/*-context.xml");
log.info("Batch Tweet Hashtag MR Job Running");
// 关闭"钩子" 为了方便在适当的时候关闭 spring ioc
// (在非web环境下,关闭spring ioc需要手动完成)
context.registerShutdownHook();
// job 发射器
// JobLaucher是一个简化的job的控制接口;基于运行时不同的标识
// 该接口并不能确保执行job是同步还是异步
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
// job
Job job = context.getBean(Job.class);
// 运行job
jobLauncher.run(job, new JobParameters());
}
}
二、xml配置文件
(1)、common 配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
// job 仓库
<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean"/>
// 事务
<bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>
// job launcher
<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher" p:jobRepository-ref="jobRepository"/>
</beans>
(2)、特殊配置
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:batch="http://www.springframework.org/schema/batch"
xmlns:hdp="http://www.springframework.org/schema/hadoop"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/hadoop http://www.springframework.org/schema/hadoop/spring-hadoop.xsd">
// 引入common 配置
<import resource="batch-common-context.xml"/>
// hdfs uri/分析目录/统计目录/分析源文件
<context:property-placeholder location="hadoop.properties"/>
<context:component-scan base-package="org.springframework.samples.hadoop.mapreduce" />
// 指定hdfs
<hdp:configuration>
fs.defaultFS=${hd.fs}
</hdp:configuration>
// 设定hdp 脚本 用于创建localSourceFile、inputDir、outputDir
<hdp:script id="setupScript" location="file-prep.groovy" run-at-startup="true">
<hdp:property name="localSourceFile" value="${localSourceFile}"/>
<hdp:property name="inputDir" value="${tweets.input.path}"/>
<hdp:property name="outputDir" value="${tweets.output.path}"/>
</hdp:script>
// 指定mapreduce step to step执行
<!-- required since Hadoop Job is a class not an interface and we need to use a Job with step scope to access #{jobParameters['...']} -->
<bean class="org.springframework.batch.core.scope.StepScope">
<property name="proxyTargetClass" value="true"/>
</bean>
// 设置job steps
<job id="job" xmlns="http://www.springframework.org/schema/batch">
<step id="hashtagcount" next="result-step">
<tasklet ref="hashtagcount-tasklet" />
</step>
<step id="result-step">
<tasklet ref="results"/>
</step>
</job>
<hdp:job-tasklet id="hashtagcount-tasklet" job-ref="hashtagcountJob" scope="step"/>
<hdp:job id="hashtagcountJob"
input-path="${tweets.input.path}"
output-path="${tweets.output.path}"
mapper="org.springframework.samples.hadoop.mapreduce.HashtagCount$TokenizerMapper"
reducer="org.springframework.samples.hadoop.mapreduce.HashtagCount$LongSumReducer"
scope="step" />
// 指定统计结果 输出
<hdp:script-tasklet id="results" scope="step">
<hdp:script location="classpath:results.groovy">
<hdp:property name="outputDir" value="${tweets.output.path}"/>
</hdp:script>
</hdp:script-tasklet>
</beans>
三、groovy脚本
// 判断分析源文件所在的目录是否存在 不存在创建 并将源文件复制到指定目录下
// 同时修改该文件夹的权限
if (!fsh.test(inputDir)) {
fsh.mkdir(inputDir);
fsh.copyFromLocal(localSourceFile, inputDir);
fsh.chmod(700, inputDir)
}
// 判断统计结果目录是否存在 存在则删除
if (fsh.test(outputDir)) {
fsh.rmr(outputDir)
}
-----------------------------------------------------------------------
// 输出分析统计结果的内容
println "RESULTS from " + outputDir
old = new File('results.txt')
if( old.exists() ) {
old.delete()
}
fsh.get(outputDir + '/part-r-*', 'results.txt');
String fileContents = new File('results.txt').text
println fileContents
以上即可完全通过xml完成mapreduce的batch处理
public class MrBatchApp {
// Log
private static final Log log = LogFactory.getLog(MrBatchApp.class);
//
public static void main(String[] args) throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
System.out.println("TEST");
// 加载对应的xml配置文件
AbstractApplicationContext context = new ClassPathXmlApplicationContext("classpath:/META-INF/spring/*-context.xml");
log.info("Batch Tweet Hashtag MR Job Running");
// 关闭"钩子" 为了方便在适当的时候关闭 spring ioc
// (在非web环境下,关闭spring ioc需要手动完成)
context.registerShutdownHook();
// job 发射器
// JobLaucher是一个简化的job的控制接口;基于运行时不同的标识
// 该接口并不能确保执行job是同步还是异步
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
// job
Job job = context.getBean(Job.class);
// 运行job
jobLauncher.run(job, new JobParameters());
}
}
二、xml配置文件
(1)、common 配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
// job 仓库
<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean"/>
// 事务
<bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>
// job launcher
<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher" p:jobRepository-ref="jobRepository"/>
</beans>
(2)、特殊配置
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:batch="http://www.springframework.org/schema/batch"
xmlns:hdp="http://www.springframework.org/schema/hadoop"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/hadoop http://www.springframework.org/schema/hadoop/spring-hadoop.xsd">
// 引入common 配置
<import resource="batch-common-context.xml"/>
// hdfs uri/分析目录/统计目录/分析源文件
<context:property-placeholder location="hadoop.properties"/>
<context:component-scan base-package="org.springframework.samples.hadoop.mapreduce" />
// 指定hdfs
<hdp:configuration>
fs.defaultFS=${hd.fs}
</hdp:configuration>
// 设定hdp 脚本 用于创建localSourceFile、inputDir、outputDir
<hdp:script id="setupScript" location="file-prep.groovy" run-at-startup="true">
<hdp:property name="localSourceFile" value="${localSourceFile}"/>
<hdp:property name="inputDir" value="${tweets.input.path}"/>
<hdp:property name="outputDir" value="${tweets.output.path}"/>
</hdp:script>
// 指定mapreduce step to step执行
<!-- required since Hadoop Job is a class not an interface and we need to use a Job with step scope to access #{jobParameters['...']} -->
<bean class="org.springframework.batch.core.scope.StepScope">
<property name="proxyTargetClass" value="true"/>
</bean>
// 设置job steps
<job id="job" xmlns="http://www.springframework.org/schema/batch">
<step id="hashtagcount" next="result-step">
<tasklet ref="hashtagcount-tasklet" />
</step>
<step id="result-step">
<tasklet ref="results"/>
</step>
</job>
<hdp:job-tasklet id="hashtagcount-tasklet" job-ref="hashtagcountJob" scope="step"/>
<hdp:job id="hashtagcountJob"
input-path="${tweets.input.path}"
output-path="${tweets.output.path}"
mapper="org.springframework.samples.hadoop.mapreduce.HashtagCount$TokenizerMapper"
reducer="org.springframework.samples.hadoop.mapreduce.HashtagCount$LongSumReducer"
scope="step" />
// 指定统计结果 输出
<hdp:script-tasklet id="results" scope="step">
<hdp:script location="classpath:results.groovy">
<hdp:property name="outputDir" value="${tweets.output.path}"/>
</hdp:script>
</hdp:script-tasklet>
</beans>
三、groovy脚本
// 判断分析源文件所在的目录是否存在 不存在创建 并将源文件复制到指定目录下
// 同时修改该文件夹的权限
if (!fsh.test(inputDir)) {
fsh.mkdir(inputDir);
fsh.copyFromLocal(localSourceFile, inputDir);
fsh.chmod(700, inputDir)
}
// 判断统计结果目录是否存在 存在则删除
if (fsh.test(outputDir)) {
fsh.rmr(outputDir)
}
-----------------------------------------------------------------------
// 输出分析统计结果的内容
println "RESULTS from " + outputDir
old = new File('results.txt')
if( old.exists() ) {
old.delete()
}
fsh.get(outputDir + '/part-r-*', 'results.txt');
String fileContents = new File('results.txt').text
println fileContents
以上即可完全通过xml完成mapreduce的batch处理
发表评论
-
nacos单机源码调试
2018-12-17 11:35 1155首先从github上获取对应的源码Nacos源码git cl ... -
jstorm源码之TransactionalState
2016-03-21 19:31 829一、作用 主要是通过结合zookeeper,在zookee ... -
jstorm源码之RotatingTransactionalState
2016-03-21 19:29 518一、作用 构建一个Rotationg transacti ... -
jstorm源码之PartitionedTridentSpoutExecutor
2016-03-21 19:28 833一、作用 Partition Spout对应的exec ... -
jstorm源码之 RichSpoutBatchExecutor
2016-03-21 19:28 0一、作用 RichSpoutBatchExecutor是IRi ... -
jstorm源码之RotatingMap
2016-03-21 19:27 819一、作用 基于LinkedList + HashM ... -
jstorm源码之 RichSpoutBatchExecutor
2016-03-21 19:24 555一、作用 RichSpoutBatchExecutor是IRi ... -
jstorm源码之TridentTopology
2016-03-16 18:12 2307在jstorm中对应TridentTopology的源码如下, ... -
jstorm操作命令
2016-03-15 18:04 2669启动ZOOPKEEPER zkServer.sh start ... -
JStorm之Supervisor简介
2016-03-15 18:02 1168一、简介Supervisor是JStorm中的工作节点,类似 ... -
JStorm介绍
2016-03-15 17:56 866一、简介Storm是开源的 ... -
mycat的使用---sqlserver和mysql
2016-01-11 14:33 8524数据库中间件mycat的使 ... -
jstorm安装
2015-12-03 19:43 1674关于jstorm单机安装可以 ... -
HBase系列一
2015-11-30 16:17 663关于hbase 一、客户端类 HTable 和 HTabl ... -
spring hadoop系列(六)---HbaseSystemException
2015-11-30 09:13 400一、源码 /** * HBase Data Access e ... -
spring hadoop系列(五)---spring hadoop hbase之HbaseSynchronizationManager
2015-11-27 18:16 832一、源码如下 /** * Synchronization m ... -
spring hadoop 系列(三)--spring hadoop hbase HbaseConfigurationFactoryBean
2015-11-27 16:28 1484一、源码分析 /** * 设定Hbase指定Configu ... -
spring hadoop 系列(二)
2015-11-27 15:26 550一、源码分析 /** * * HbaseAccesso ... -
spring hadoop之mapreduce batch
2015-11-24 15:51 571一、测试 // 定义hadoop configuration ... -
spring hadoop
2015-11-24 10:19 547一、源码 // 如下代码实现 读取指定hdfs路径下tmp文件 ...
相关推荐
第二章(Hadoop大数据处理实战)搭建Hadoop分布式集群.pdf第二章(Hadoop大数据处理实战)搭建Hadoop分布式集群.pdf第二章(Hadoop大数据处理实战)搭建Hadoop分布式集群.pdf第二章(Hadoop大数据处理实战)搭建Hadoop分布式...
第四章(Hadoop大数据处理实战)Hadoop分布式文件系统.pdf第四章(Hadoop大数据处理实战)Hadoop分布式文件系统.pdf第四章(Hadoop大数据处理实战)Hadoop分布式文件系统.pdf第四章(Hadoop大数据处理实战)Hadoop分布式文件...
集成spring3-hadoop0.2.0 eclipse3.7快速开发 mapreduce程序!
Spring for Apache Hadoop 提供了 Spring 框架用于创建和运行 Hadoop MapReduce、Hive 和 Pig 作业的功能,包括 HDFS 和 HBase。如果你需要简单的基于 Hadoop 进行作业调度,你可添加 Spring for Apache Hadoop 命名...
第五章(Hadoop大数据处理实战)Hadoop的IO操作.pdf第五章(Hadoop大数据处理实战)Hadoop的IO操作.pdf第五章(Hadoop大数据处理实战)Hadoop的IO操作.pdf第五章(Hadoop大数据处理实战)Hadoop的IO操作.pdf第五章(Hadoop大...
《Hadoop大数据处理》共10章涉及的主题包括大数据处理概论、基于Hadoop的大数据处理框架、MapReduce计算模式、使用HDFS存储大数据、HBase大数据库、大数据的分析处理、Hadoop环境下的数据整合、Hadoop集群的管理与...
利用hadoop的rpc包简单与spring结合,实现了一个简单的分布式。
Spring Data for Apache Hadoop API。 Spring Data for Apache Hadoop 开发文档
spring data hadoop 参考文档
Hadoop大数据处理技术基础与实践(第2版)(微课版)PPT-课件.rar
本书介绍了Hadoop技术的相关知识,并将理论知识与实际项目相结合。全书共分为三个部分:基础篇、应用篇和总结篇。
Hadoop海量数据处理 Hadoop海量数据处理 Hadoop海量数据处理
Hadoop海量文本处理2
spring hadoop 分布式 海量数据 集群 mapreduce 将hadoop于spring集成到一起工作
hadoop海量数据处理技术详解,包括hdfs、MapReduce、hive、sqoop等相关技术和伪代码,代码是使用python语言写的。
本书从Hadoop的缘起开始,由浅入深,结合理论和实践,全方位地介绍Hadoop这一高性能处理海量数据集的理想工具。全书共16章,3个附录,涉及的主题包括:Haddoop简介;MapReduce简介;Hadoop分布式文件系统;Hadoop...
hadoop海量数据处理.pdfhadoop海量数据处理.pdfhadoop海量数据处理.pdfhadoop海量数据处理.pdfhadoop海量数据处理.pdfhadoop海量数据处理.pdf
Hadoop大数据处理讲义-C1.海量数据处理概论
java整合spring和hadoop HDFS全部jar
Hadoop海量文本处理1