hadoop java编程,hadoop怎么运行java代码

2022-07-31 09:16:34 分类：java 阅读()

　　hadoop java编程,hadoop怎么运行java代码

　　

目录

实验环境实验内容测试爪哇远程连接大数据

 

　　

实验环境

hadoop版本：3.3.2jdk版本：1.8hadoop安装系统：ubuntu18.04编程环境：想法编程主机：windows

 

　　

实验内容

 

　　

测试Java远程连接hadoop

创建maven工程，引入以下依赖：

 

　　依赖关系groupIdorg.testng/groupId artifactIdtestng/artifactId版本发布/版本范围编译/范围/依赖关系依赖关系groupIdorg.apache.hadoop/groupId artifact id Hadoop-通用/artifact id/依赖项依赖关系groupIdorg.apache.hadoop/groupId artifact id Hadoop-HDFS/artifact id版本3。3 .2/版本/依赖关系依赖关系groupIdorg.apache.hadoop/groupId artifact id Hadoop-common/artifact id版本3。3 .2/版本/依赖关系依赖关系groupIdorg.apache.hadoop/groupId artifact id Hadoop-核心/artifactId版本1。2 .1/版本/相关性虚拟机的/etc/hosts配置

　　hdfs-site.xml配置

　　配置属性namedfs.replication/name价值1/价值/属性属性namedfs.namenode.name.dir/name值文件：/root/rDesk/Hadoop-3。3 .2/tmp/DFS/名称/值/属性属性namedfs.datanode.http.address/name价值VM-12-11-Ubuntu :50010/价值/财产属性命名DFS。客户。使用。datanode。马血清

　　tname</name> <value>true</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>file:/root/rDesk/hadoop-3.3.2/tmp/dfs/data</value> </property></configuration>

 

　　core-site.xml配置

　　

<configuration> <property> <name>hadoop.tmp.dir</name> <value>file:/root/rDesk/hadoop-3.3.2/tmp</value> <description>Abase for other temporary directories.</description> </property> <property> <name>fs.defaultFS</name> <value>hdfs://VM-12-11-ubuntu:9000</value> </property></configuration>

 

　　启动hadoop

　　

sbin/start-dfs.sh

 

　　

主机的hosts（C:WindowsSystem32driversetc）文件配置

 

　　

 

　　尝试连接到虚拟机的hadoop并读取文件内容，这里我读取hdfs下的/root/iinput文件内容

　　

 

　　Java代码：

　　

import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FSDataInputStream;import org.apache.hadoop.fs.FileStatus;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.hdfs.DistributedFileSystem;public class TestConnectHadoop { public static void main(String[] args) throws Exception { String hostname = "VM-12-11-ubuntu"; String HDFS_PATH = "hdfs://" + hostname + ":9000"; Configuration conf = new Configuration(); conf.set("fs.defaultFS", HDFS_PATH); conf.set("fs.hdfs.impl", DistributedFileSystem.class.getName()); conf.set("dfs.client.use.datanode.hostname", "true"); FileSystem fs = FileSystem.get(conf); FileStatus[] fileStatuses = fs.listStatus(new Path("/")); for (FileStatus fileStatus : fileStatuses) { System.out.println(fileStatus.toString()); } FileStatus fileStatus = fs.getFileStatus(new Path("/root/iinput")); System.out.println(fileStatus.getOwner()); System.out.println(fileStatus.getGroup()); System.out.println(fileStatus.getPath()); FSDataInputStream open = fs.open(fileStatus.getPath()); byte[] buf = new byte[1024]; int n = -1; StringBuilder sb = new StringBuilder(); while ((n = open.read(buf)) > 0) { sb.append(new String(buf, 0, n)); } System.out.println(sb); }}

运行结果：

 

　　

 

　　编程实现一个类MyFSDataInputStream，该类继承org.apache.hadoop.fs.FSDataInputStream",要求如下： ①实现按行读取HDFS中指定文件的方法readLine（），如果读到文件末尾，则返回为空，否则返回文件一行的文本

　　思路：emmm我的思路比较简单，只适用于该要求，仅作参考。将所有的数据读取出来存储起来，然后根据换行符进行拆分，将拆分的字符串数组存储起来，用于readline返回

　　Java代码

　　

import org.apache.hadoop.fs.FSDataInputStream;import java.io.IOException;import java.io.InputStream;public class MyFSDataInputStream extends FSDataInputStream { private String data = null; private String[] lines = null; private int count = 0; private FSDataInputStream in; public MyFSDataInputStream(InputStream in) throws IOException { super(in); this.in = (FSDataInputStream) in; init(); } private void init() throws IOException { byte[] buf = new byte[1024]; int n = -1; StringBuilder sb = new StringBuilder(); while ((n = this.in.read(buf)) > 0) { sb.append(new String(buf, 0, n)); } data = sb.toString(); lines = data.split("n"); } /** * 实现按行读取HDFS中指定文件的方法”readLine（）“，如果读到文件末尾，则返回为空，否则返回文件一行的文本 */ public String read_line() { return count < lines.length ? lines[count++] : null; }}

测试类：

 

　　

import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FSDataInputStream;import org.apache.hadoop.fs.FileStatus;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.hdfs.DistributedFileSystem;public class TestConnectHadoop { public static void main(String[] args) throws Exception { String hostname = "VM-12-11-ubuntu"; String HDFS_PATH = "hdfs://" + hostname + ":9000"; Configuration conf = new Configuration(); conf.set("fs.defaultFS", HDFS_PATH); conf.set("fs.hdfs.impl", DistributedFileSystem.class.getName()); conf.set("dfs.client.use.datanode.hostname", "true"); FileSystem fs = FileSystem.get(conf); FileStatus fileStatus = fs.getFileStatus(new Path("/root/iinput")); System.out.println(fileStatus.getOwner()); System.out.println(fileStatus.getGroup()); System.out.println(fileStatus.getPath()); FSDataInputStream open = fs.open(fileStatus.getPath()); MyFSDataInputStream myFSDataInputStream = new MyFSDataInputStream(open); String line = null; int count = 0; while ((line = myFSDataInputStream.read_line()) != null ) { System.out.printf("line %d is: %sn", count++, line); } System.out.println("end"); }}

运行结果：

 

　　

 

　　②实现缓存功能，即利用MyFSDataInputStream读取若干字节数据时，首先查找缓存，如果缓存中有所需要数据，则直接由缓存提供，否则从HDFS中读取数据

　　

import org.apache.hadoop.fs.FSDataInputStream;import java.io.BufferedInputStream;import java.io.IOException;import java.io.InputStream;public class MyFSDataInputStream extends FSDataInputStream { private BufferedInputStream buffer; private String[] lines = null; private int count = 0; private FSDataInputStream in; public MyFSDataInputStream(InputStream in) throws IOException { super(in); this.in = (FSDataInputStream) in; init(); } private void init() throws IOException { byte[] buf = new byte[1024]; int n = -1; StringBuilder sb = new StringBuilder(); while ((n = this.in.read(buf)) > 0) { sb.append(new String(buf, 0, n)); } //缓存数据读取 buffer = new BufferedInputStream(this.in); lines = sb.toString().split("n"); } /** * 实现按行读取HDFS中指定文件的方法”readLine（）“，如果读到文件末尾，则返回为空，否则返回文件一行的文本 */ public String read_line() { return count < lines.length ? lines[count++] : null; } @Override public int read() throws IOException { return this.buffer.read(); } public int readWithBuf(byte[] buf, int offset, int len) throws IOException { return this.buffer.read(buf, offset, len); } public int readWithBuf(byte[] buf) throws IOException { return this.buffer.read(buf); }}

到此这篇关于利用Java连接Hadoop进行编程的文章就介绍到这了,更多相关Java连接Hadoop内容请搜索盛行IT以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT！

 

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。

标签：

相关文章阅读

  • java中swing组件,javaswing常用控件,Java编程swing组件JLabel详解以及使用示例
  • ,,Java编程实现数组转成list及list转数组的方法
  • ,,java编程abstract类和方法详解
  • 编程时为什么要空一行,java编程填空题
  • 在java编程中swing包中的组件处理事件时,swing的事件处理机制
  • 在java编程中swing包中的组件处理事件时,swing的事件处理机制_1
  • java控制语句案例,Java编程的逻辑
  • 学java编程可以做什么,想学java编程
  • java编程中找不到符号是什么原因,java 错误- 找不到符号遇到的几种情况
  • java编程思想适合入门吗,java编程思想讲的是什么
  • 编译程序的顺序,java编程顺序
  • java编程排序算法实现,十大经典排序算法java
  • 扫雷游戏java编程,java扫雷代码简单实现
  • Java开发编辑器,java编程开发工具
  • java汉诺塔问题编程,汉诺塔java编程讲解
    • 留言与评论（共有 条评论）
       
    验证码：