柚子快報(bào)激活碼778899分享：hdfs hadoop

http://yzkb.51969.com/

上傳文件

好的，你需要打開虛擬機(jī)終端并輸入以下命令：

1. 查看當(dāng)前目錄下的文件和文件夾：

```

ls

```

2. 進(jìn)入某個(gè)文件夾：

```

cd 文件夾名稱

```

例如，如果你要進(jìn)入名為"bigdata"的文件夾，可以輸入：

```

cd bigdata

```

3. 查看當(dāng)前所在的目錄：

```

pwd

```

4. 上傳文件到HDFS：

- 使用hadoop fs命令上傳文件：

```

hadoop fs -put 本地文件路徑 HDFS文件路徑

```

例如，如果你要將本地目錄下的"file.txt"文件上傳到HDFS的"/user/hadoop/"目錄下，可以輸入：

```

hadoop fs -put file.txt /user/hadoop/

```

- 使用Java API上傳文件：

首先需要編寫Java代碼，具體代碼實(shí)現(xiàn)可以參考Hadoop官方文檔。然后通過終端進(jìn)入Java代碼所在的目錄，然后編譯代碼并執(zhí)行。

```

javac -classpath $(hadoop classpath) Java文件名.java

java -classpath $(hadoop classpath):. Java文件名

```

- 使用HDFS API上傳文件：

同樣需要編寫Java代碼實(shí)現(xiàn)文件上傳，具體實(shí)現(xiàn)可以參考Hadoop官方文檔。然后通過終端進(jìn)入Java代碼所在的目錄，然后編譯代碼并執(zhí)行。

```

javac -classpath $(hadoop classpath) Java文件名.java

java -classpath $(hadoop classpath):. Java文件名

```

按某字段降序并取Top10

好的，下面是從新建項(xiàng)目開始的具體流程：

1. 在IntelliJ IDEA中新建一個(gè)Maven項(xiàng)目，選擇模板"quickstart"，填寫項(xiàng)目名稱和項(xiàng)目路徑，然后點(diǎn)擊"Finish"按鈕。

2. 在項(xiàng)目的pom.xml文件中添加Hadoop的依賴項(xiàng)，具體代碼如下：

```

????org.apache.hadoop

????hadoop-common

????3.3.0

????org.apache.hadoop

????hadoop-hdfs

????3.3.0

```

3. 在項(xiàng)目中創(chuàng)建一個(gè)Java類，例如"TopTen.java"，并在類中編寫代碼，具體代碼如下：

```java

import java.io.IOException;

import java.util.*;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.*;

import org.apache.hadoop.mapreduce.*;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class TopTen {

????public static class TopTenMapper extends Mapper {

????????private TreeMap topTen = new TreeMap();

????????@Override

????????public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

????????????String line = value.toString();

????????????String[] fields = line.split(",");

????????????if (fields.length != 2) {

????????????????return;

????????????}

????????????String word = fields[0];

????????????int count = Integer.parseInt(fields[1]);

????????????topTen.put(count, word);

????????????if (topTen.size() > 10) {

????????????????topTen.remove(topTen.firstKey());

????????????}

????????}

????????@Override

????????protected void cleanup(Context context) throws IOException, InterruptedException {

????????????for (Map.Entry entry : topTen.entrySet()) {

????????????????context.write(new Text(entry.getValue()), new IntWritable(entry.getKey()));

????????????}

????????}

????}

????public static class TopTenReducer extends Reducer {

????????private TreeMap topTen = new TreeMap();

????????@Override

????????public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

????????????int sum = 0;

????????????for (IntWritable value : values) {

????????????????sum += value.get();

????????????}

????????????topTen.put(sum, key.toString());

????????????if (topTen.size() > 10) {

????????????????topTen.remove(topTen.firstKey());

????????????}

????????}

????????@Override

????????protected void cleanup(Context context) throws IOException, InterruptedException {

????????????for (Map.Entry entry : topTen.descendingMap().entrySet()) {

????????????????context.write(new Text(entry.getValue()), new IntWritable(entry.getKey()));

????????????}

????????}

????}

????public static void main(String[] args) throws Exception {

????????Configuration conf = new Configuration();

????????Job job = Job.getInstance(conf, "TopTen");

????????job.setJarByClass(TopTen.class);

????????job.setMapperClass(TopTenMapper.class);

????????job.setReducerClass(TopTenReducer.class);

????????job.setOutputKeyClass(Text.class);

????????job.setOutputValueClass(IntWritable.class);

????????job.setInputFormatClass(TextInputFormat.class);

????????job.setOutputFormatClass(TextOutputFormat.class);

????????FileInputFormat.addInputPath(job, new Path(args[0]));

????????FileOutputFormat.setOutputPath(job, new Path(args[1]));

????????System.exit(job.waitForCompletion(true) ? 0 : 1);

????}

}

```

4. 在終端中使用以下命令將數(shù)據(jù)文件上傳到HDFS：

```

hadoop fs -put 數(shù)據(jù)文件路徑 HDFS文件路徑

```

例如，如果你要將本地目錄下的"data.txt"文件上傳到HDFS的"/user/hadoop/"目錄下，可以輸入：

```

hadoop fs -put data.txt /user/hadoop/

```

5. 在終端中使用以下命令運(yùn)行MapReduce作業(yè)：

```

hadoop jar 項(xiàng)目JAR包路徑 數(shù)據(jù)文件HDFS路徑 作業(yè)輸出HDFS路徑

```

例如，如果你要運(yùn)行名為"TopTen"的作業(yè)，數(shù)據(jù)文件在"/user/hadoop/data.txt"，作業(yè)輸出路徑為"/user/hadoop/output"，可以輸入：

```

hadoop jar TopTen.jar /user/hadoop/data.txt /user/hadoop/output

```

6. 在終端中使用以下命令查看輸出結(jié)果：

```

hadoop fs -cat 作業(yè)輸出HDFS路徑/part-r-00000

```

例如，如果你要查看作業(yè)輸出結(jié)果的前10行，可以輸入：

```

hadoop fs -cat /user/hadoop/output/part-r-00000 | head -n 10

```

代碼解釋：

- TopTenMapper類繼承了Mapper類，并實(shí)現(xiàn)了map方法。在map方法中，首先將輸入的每一行數(shù)據(jù)按照逗號(hào)分隔為單詞和出現(xiàn)次數(shù)兩個(gè)部分，然后將單詞作為key，出現(xiàn)次數(shù)作為value加入到一個(gè)TreeMap中。如果TreeMap的大小超過10，則移除最小的元素。在cleanup方法中，將TreeMap中的每個(gè)元素寫入到Context中。

- TopTenReducer類繼承了Reducer類，并實(shí)現(xiàn)了reduce方法。在reduce方法中，首先將輸入的每個(gè)key-value對(duì)中的value累加起來作為sum，然后將key作為value，sum作為key加入到一個(gè)TreeMap中。如果TreeMap的大小超過10，則移除最小的元素。在cleanup方法中，將TreeMap中的每個(gè)元素按照降序?qū)懭氲紺ontext中。

- main方法中創(chuàng)建了一個(gè)Job對(duì)象，并設(shè)置了相關(guān)的屬性，包括作業(yè)名稱、Jar包類、Mapper類、Reducer類、輸出key類型、輸出value類型、輸入格式、輸出格式、輸入路徑、輸出路徑等。最后提交作業(yè)并等待作業(yè)完成。

注意：在實(shí)際操作中，需要將代碼中的輸入路徑和輸出路徑修改為實(shí)際的路徑。另外，數(shù)據(jù)文件的格式應(yīng)該與代碼中的格式一致。

分組求平均數(shù)(分組條件涉及兩個(gè)字段 有多種實(shí)現(xiàn)方式 比如一個(gè)字段作為key，另一個(gè)字段作為分區(qū)的條件;或者兩個(gè)字段聯(lián)合作為key，都可以)

好的，以下是一個(gè)新建項(xiàng)目并實(shí)現(xiàn)分組求平均數(shù)的步驟：

1. 打開IDE（比如Eclipse、IntelliJ IDEA等），創(chuàng)建一個(gè)新的Java項(xiàng)目。

2. 在項(xiàng)目中創(chuàng)建一個(gè)新的Java類，命名為MyMapper，并繼承自org.apache.hadoop.mapreduce.Mapper。

???```java

???import org.apache.hadoop.io.LongWritable;

???import org.apache.hadoop.io.Text;

???import org.apache.hadoop.mapreduce.Mapper;

???public class MyMapper extends Mapper {

???????public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

???????????// TODO: 實(shí)現(xiàn)map方法

???????}

???}

???```

3. 在MyMapper類中實(shí)現(xiàn)map方法，該方法接收三個(gè)參數(shù)：一個(gè)LongWritable類型的參數(shù)表示輸入的key，一個(gè)Text類型的參數(shù)表示輸入的value，一個(gè)Context類型的參數(shù)用于輸出結(jié)果。在map方法中，我們需要對(duì)輸入的value進(jìn)行解析，然后將兩個(gè)字段作為輸出的key，將一個(gè)字段作為輸出的value，輸出到Context中。以下是一個(gè)簡(jiǎn)單的例子：

???```java

???public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

???????String line = value.toString();

???????String[] fields = line.split(",");

???????String field1 = fields[0];

???????String field2 = fields[1];

???????String field3 = fields[2];

???????context.write(new Text(field1 + "," + field2), new Text(field3 + ",1"));

???}

???```

???這個(gè)例子中，我們假設(shè)輸入的value是一個(gè)逗號(hào)分隔的三個(gè)字段，第一個(gè)和第二個(gè)字段是分組條件，第三個(gè)字段是要求平均數(shù)的值。我們首先將前兩個(gè)字段拼接成一個(gè)字符串作為輸出的key，將第三個(gè)字段和1拼接成一個(gè)字符串作為輸出的value，輸出到Context中。

4. 創(chuàng)建一個(gè)新的Java類，命名為MyReducer，并繼承自org.apache.hadoop.mapreduce.Reducer。

???```java

???import org.apache.hadoop.io.LongWritable;

???import org.apache.hadoop.io.Text;

???import org.apache.hadoop.mapreduce.Reducer;

???public class MyReducer extends Reducer {

???????public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

???????????// TODO: 實(shí)現(xiàn)reduce方法

???????}

???}

???```

5. 在MyReducer類中實(shí)現(xiàn)reduce方法，該方法接收三個(gè)參數(shù)：一個(gè)Text類型的參數(shù)表示輸入的key，一個(gè)Iterable類型的參數(shù)表示輸入的value列表，一個(gè)Context類型的參數(shù)用于輸出結(jié)果。在reduce方法中，我們需要將輸入的value列表中的所有值相加，然后求平均數(shù)，最后將結(jié)果輸出到Context中。以下是一個(gè)簡(jiǎn)單的例子：

???```java

???public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

???????long sum = 0;

???????long count = 0;

???????for (Text value : values) {

???????????String[] fields = value.toString().split(",");

???????????sum += Long.parseLong(fields[0]);

???????????count += Long.parseLong(fields[1]);

???????}

???????double avg = (double)sum / count;

???????context.write(key, new DoubleWritable(avg));

???}

???```

???這個(gè)例子中，我們將輸入的value列表中的所有值相加，然后求平均數(shù)，最后將結(jié)果輸出到Context中。

6. 在項(xiàng)目中創(chuàng)建一個(gè)新的Java類，命名為MyJob。

???```java

???import org.apache.hadoop.conf.Configuration;

???import org.apache.hadoop.fs.Path;

???import org.apache.hadoop.io.DoubleWritable;

???import org.apache.hadoop.io.LongWritable;

???import org.apache.hadoop.io.Text;

???import org.apache.hadoop.mapreduce.Job;

???import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

???import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

???import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

???public class MyJob {

???????public static void main(String[] args) throws Exception {

???????????Configuration conf = new Configuration();

???????????Job job = Job.getInstance(conf, "MyJob");

???????????job.setJarByClass(MyJob.class);

???????????job.setMapperClass(MyMapper.class);

???????????job.setReducerClass(MyReducer.class);

???????????job.setOutputKeyClass(Text.class);

???????????job.setOutputValueClass(Text.class);

???????????job.setInputFormatClass(TextInputFormat.class);

???????????job.setOutputFormatClass(TextOutputFormat.class);

???????????FileInputFormat.addInputPath(job, new Path(args[0]));

???????????FileOutputFormat.setOutputPath(job, new Path(args[1]));

???????????System.exit(job.waitForCompletion(true) ? 0 : 1);

???????}

???}

???```

???在MyJob類中，我們創(chuàng)建了一個(gè)新的Job實(shí)例，設(shè)置了該Job的輸入路徑、輸出路徑、Mapper類、Reducer類等屬性，然后將該Job提交到Hadoop集群中運(yùn)行。

7. 最后，我們可以在命令行中執(zhí)行以下命令來運(yùn)行我們的程序：

???```bash

???hadoop jar MyProject.jar MyJob /path/to/input /path/to/output

???```

???其中，MyProject.jar是我們打包后的程序包文件名，MyJob是我們的Job類名，/path/to/input和/path/to/output分別是輸入和輸出的HDFS路徑。

柚子快報(bào)激活碼778899分享：hdfs hadoop

http://yzkb.51969.com/

好文推薦