Bu örnek, bir JAR paketini ve metin dosyasını MaxCompute kaynağı olarak nasıl kayıt altına alacağınızı ve bunları bir tabloya veri yüklemek için MapReduce işinde nasıl kullanacağınızı gösterir.
Çalışma mantığı
Mapper, kurulum sırasında context.readResourceFileAsStream() kullanarak metin dosyasını okur, her satırı ayrıştırır ve kayıtları çıktı tablosuna yazar. İşin reduce aşaması yoktur.
-resources bayrağı ile -classpath bayrağının amaçları farklıdır:
|
|
|
|
|
|
|
|
|
Ön koşullar
Başlamadan önce aşağıdakileri tamamladığınızdan emin olun:
Hızlı başlangıç bölümünde açıklanan ortam kurulumunu tamamlayın
Yerel MaxCompute kurulumunuzun
bin\data\resourcesdizinindemapreduce-examples.jardosyasının bulunduğundan emin olun
Tabloları ve kaynakları hazırlama
Aşağıdaki komutları MaxCompute istemcisinde çalıştırın.
-
Çıktı tablosunu oluşturun:
CREATE TABLE mr_upload_src(key BIGINT, value STRING); -
Metin dosyasını ve JAR paketini kaynak olarak ekleyin:
add file data\resources\import.txt -f; add jar data\resources\mapreduce-examples.jar -f;JAR paketini ilk kez eklerken
-fbayrağını yok sayabilirsiniz.import.txtdosyası aşağıdaki verileri içerir:1000,odps
İşi çalıştırma
Yükleme işini başlatmak için MaxCompute istemcisinde aşağıdaki komutu çalıştırın:
jar -resources mapreduce-examples.jar,import.txt -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.Upload import.txt mr_upload_src;
Parametre referansı:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Sonucu doğrulama
İş tamamlandıktan sonra mr_upload_src tablosunu sorgulayın. Tablo, import.txt dosyasındaki 1000,odps satırıyla doğrudan eşleşen aşağıdaki satırı içermelidir (0. alan → key, 1. alan → value):
+------------+------------+
| key | value |
+------------+------------+
| 1000 | odps |
+------------+------------+
Örnek kod
Project Object Model (POM) bağımlılık yapılandırması için Hızlı başlangıç bölümüne bakın.
package com.aliyun.odps.mapred.open.example;
import java.io.BufferedInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.TableInfo;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.TaskContext;
import com.aliyun.odps.mapred.conf.JobConf;
import com.aliyun.odps.mapred.utils.InputUtils;
import com.aliyun.odps.mapred.utils.OutputUtils;
import com.aliyun.odps.mapred.utils.SchemaUtils;
/**
* Upload
* Import data from a text file into a table.
*/
public class Upload {
public static class UploadMapper extends MapperBase {
@Override
public void setup(TaskContext context) throws IOException {
Record record = context.createOutputRecord();
StringBuilder importdata = new StringBuilder();
BufferedInputStream bufferedInput = null;
try {
byte[] buffer = new byte[1024];
int bytesRead = 0;
// Get the resource file name from JobConf
String filename = context.getJobConf().get("import.filename");
// Read the resource file distributed to this worker node
bufferedInput = context.readResourceFileAsStream(filename);
while ((bytesRead = bufferedInput.read(buffer)) != -1) {
String chunk = new String(buffer, 0, bytesRead);
importdata.append(chunk);
}
// Parse each line: split by comma, write key (BIGINT) and value (STRING)
String lines[] = importdata.toString().split("\n");
for (int i = 0; i < lines.length; i++) {
String[] ss = lines[i].split(",");
record.set(0, Long.parseLong(ss[0].trim()));
record.set(1, ss[1].trim());
context.write(record);
}
} catch (FileNotFoundException ex) {
throw new IOException(ex);
} catch (IOException ex) {
throw new IOException(ex);
} finally {
}
}
@Override
public void map(long recordNum, Record record, TaskContext context)
throws IOException {
}
}
public static void main(String[] args) throws Exception {
if (args.length != 2) {
System.err.println("Usage: Upload <import_txt> <out_table>");
System.exit(2);
}
JobConf job = new JobConf();
job.setMapperClass(UploadMapper.class);
// Pass the resource file name to the mapper via JobConf
job.set("import.filename", args[0]);
// Set reducers to 0: this is a map-only job
job.setNumReduceTasks(0);
job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint"));
job.setMapOutputValueSchema(SchemaUtils.fromString("value:string"));
InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), job);
OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
JobClient.runJob(job);
}
}
JobConf ayarlama
Yukarıdaki örnekte, SDK'daki JobConf arabirimi kullanılarak iş özellikleri ayarlanmıştır. Alternatif olarak, jar komutundaki -conf parametresini kullanarak bir JobConf yapılandırma dosyası belirtebilirsiniz:
jar -conf <your-jobconf-file> ...