软件编程
位置：首页>> 软件编程>> java编程>> 利用Java连接Hadoop进行编程

利用Java连接Hadoop进行编程

作者：wr456wr　　发布时间：2022-11-12 09:02:12　

标签：Java,连接,Hadoop,编程

实验环境

  • hadoop版本：3.3.2

  • jdk版本：1.8

  • hadoop安装系统：ubuntu18.04

  • 编程环境：IDEA

  • 编程主机：windows

实验内容

测试Java远程连接hadoop

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

<dependency>
               <groupId>org.testng</groupId>
               <artifactId>testng</artifactId>
               <version>RELEASE</version>
               <scope>compile</scope>
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-hdfs</artifactId>
               <version>3.3.2</version>
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>3.3.2</version>
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-core</artifactId>
               <version>1.2.1</version>
           </dependency>

虚拟机的/etc/hosts配置

利用Java连接Hadoop进行编程

hdfs-site.xml配置

<configuration>
       <property>
               <name>dfs.replication</name>
               <value>1</value>
       </property>
       <property>
               <name>dfs.namenode.name.dir</name>
               <value>file:/root/rDesk/hadoop-3.3.2/tmp/dfs/name</value>
       </property>
       <property>
               <name>dfs.datanode.http.address</name>
               <value>VM-12-11-ubuntu:50010</value>
       </property>
       <property>
               <name>dfs.client.use.datanode.hostname</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>

利用Java连接Hadoop进行编程

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>

利用Java连接Hadoop进行编程

启动hadoop

sbin/start-dfs.sh

主机的hosts（C:\Windows\System32\drivers\etc）文件配置

利用Java连接Hadoop进行编程

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

利用Java连接Hadoop进行编程

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);
   }
}

运行结果：

利用Java连接Hadoop进行编程

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

思路：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: ％s\n", count++, line);
       }
       System.out.println("end");

}
}

运行结果：

利用Java连接Hadoop进行编程

②实现缓存功能，即利用&rdquo;MyFSDataInputStream&ldquo;读取若干字节数据时，首先查找缓存，如果缓存中有所需要数据，则直接由缓存提供，否则从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);
   }
}

来源：https://blog.csdn.net/wr456wr/article/details/125363436

0
投稿

猜你喜欢

  • 概述：Flutter 标签类控件大全ChipFlutter内置了多个标签类控件，但本质上它们都是同一个控件，只不过是属性参数不同而已，在学习
  • anroid 5.0 Design  v7 包中引用了TabLayout 简单快速的写出属于自己的Tab切换效果 如图所示：但是正
  •  关于java的字符串处理我们一般使用String类和StringBuffer类那么String类和StringBuffer类的区
  • 遇到一个@ConditionalOnMissingBean失效的问题，今天花点时间来分析一下。现场回放services首先介绍下代码结构：有
  • 以下实例演示了如何使用 retainAll () 方法来计算两个数组的交集：Main.java 文件：import java.util.Ar
  • 本文以实例详述了C#实现Socket通信的解决方法，具体实现步骤如下：1、首先打开VS新建两个控制台应用程序：ConsoleApplicat
  • 要“监听”事件，我们总是可以将“ * ”作为事件源中的另一个方法写入事件，但这将使事件源与 * 的逻辑紧密耦合。对于实际事件，我们比直接方法
  • 最近开发的时候，偶尔遇到在线上稳定运行的webview内嵌的h5页面加载不出来，一直定位不到具体原因（因为我们自己做的兼容性测试上不重现），
  • 前两年写的东西，现在整理一下发出来！以前公司需要做WebService，并且对WebService的SoapHeader进行加密，所以就写了
  • 本文实例为大家分享了Android实现Window弹窗效果的具体代码，供大家参考，具体内容如下效果图第一步 准备弹窗的布局，新建XML文件
  • 一、反射概述反射机制指的是Java在运行时候有一种自观的能力，能够了解自身的情况为下一步做准备，其想表达的意思就是：在运行状态中，对于任意一
  • 概述：堆排序是利用构建“堆”的方法确定具有最大值的数据元素，并把该元素与最后位置上的元素交换。可将任意一个由n个数据元素构成的序列按照（a1
  • PS:本文使用jdk1.7解析1.Object类 的equals 方法   /**   &
  • 执行如下的jni调用：package jni;public class JNITransObject { public nativ
  • 背景最近遇到一个有意思的事情，java应用运行在阿里云的ack集群中，某一天有个应用启动突然发现阿里云上的agent都没有注册了，于是开始排
  • 概述企业内部一般都有一套单点登录系统（常用的实现有apereo cas），所有的内部系统的登录认证都对接它。本文介绍spring boot的
  • 项目上线之后，如果日志打印的很模糊或者业务逻辑比较复杂，有时候无法定位具体的错误原因，因此可以通过IDEA远程代理进行Debug。线上的代码
  • 蓝牙，时下最流行的智能设备传输数据的方式之一，通过手机app和智能设备进行连接，获取设备上的测量数据，我们生活中随处可见的比如蓝牙智能手环，
  • 本文实例为大家分享了C++实现希尔排序的具体代码，供大家参考，具体内容如下一、思路：希尔排序:又称缩小增量排序，是一种改进的插入排序算法，是
  • 前言作为Java开发者，我们每天都会创建大量的对象，但是，我们总是使用管理依赖系统（如Spring框架）来创建这些对象。其实还有其他方法可以
手机版 软件编程 asp之家 www.aspxhome.com