注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。
ここで紹介する例には、より高度な内容が含まれています。このセクションを読む前に、トランスフォームの定義に関するセクションを確認してください。
このページでは、Transforms Java を使用したさまざまなデータ変換の例を紹介します。
ここで紹介する例は、ファイルを扱うデータ変換です。変換処理でファイルにアクセスするには、低レベルの Transform を定義する必要があります。これは、基盤となるデータセットファイルが FoundryInput および FoundryOutput オブジェクトを通じて公開されるためです。低レベルのトランスフォームは、高レベルのトランスフォームとは異なり、compute 関数の入力が FoundryInput 型、出力が FoundryOutput 型であることを前提としています。ここに掲載する例も、手動登録を想定した低レベルのトランスフォームです。
この例では、.zip ファイルを入力として受け取ります。ファイルを解凍し、出力 FileSystem に書き込みます。.zip は分割できないため、この処理は Spark を使用して .zip ファイル単位で並列に実行されます。単一の圧縮ファイル内で解凍処理を並列化するには、.bz2 などの分割可能なファイル形式を使用します。各出力パスには、入力 .zip ファイルの論理パスがプレフィックスとして付加されます。同じ出力パスへの書き込みは、エグゼキューターごとに1回しかできません。エントリー名だけからパスを生成すると、2つのアーカイブに同じ名前のエントリーが含まれている場合、Attempted to produce the same output twice というエラーで失敗する可能性があります。競合する書き込みが異なるエグゼキューターで発生した場合、エラーは発生せず、生成される出力は未定義になります。
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47/* * (c) Copyright 2018 Palantir Technologies Inc. すべての権利を留保します。 */ package com.palantir.transforms.java.examples; import com.google.common.io.ByteStreams; import com.palantir.transforms.lang.java.api.Compute; import com.palantir.transforms.lang.java.api.FoundryInput; import com.palantir.transforms.lang.java.api.FoundryOutput; import com.palantir.transforms.lang.java.api.ReadOnlyLogicalFileSystem; import com.palantir.transforms.lang.java.api.WriteOnlyLogicalFileSystem; import com.palantir.util.syntacticpath.Paths; import java.io.IOException; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; /** * Spark を使ってファイルを並列に解凍する例です。 * <p> * 処理はエグゼキューターに分散されます。 */ public final class UnzipWithSpark { @Compute public void compute(FoundryInput zipFiles, FoundryOutput output) throws IOException { ReadOnlyLogicalFileSystem inputFileSystem = zipFiles.asFiles().getFileSystem(); WriteOnlyLogicalFileSystem outputFileSystem = output.getFileSystem(); inputFileSystem.filesAsDataset().foreach(portableFile -> { // 「processWith」は、指定された入力ファイルの InputStream を提供します。 portableFile.processWithThenClose(stream -> { try (ZipInputStream zis = new ZipInputStream(stream)) { ZipEntry entry; // .zip ファイル内の各ファイルを出力ファイルシステムに書き込みます。 while ((entry = zis.getNextEntry()) != null) { outputFileSystem.writeTo( Paths.get(portableFile.getLogicalPath().toString(), entry.getName()), outputStream -> ByteStreams.copy(zis, outputStream)); } return null; } catch (IOException e) { throw new RuntimeException(e); } }); }); } }
この例では、.csv、.gz、.zip ファイルを入力として受け取ります。ファイルを解凍し、出力 DataFrame に書き込みます。この処理は Spark を使用して並列に実行されます。
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138/* * (c) Copyright 2018 Palantir Technologies Inc. すべての権利を留保します。 */ package com.palantir.transforms.java.examples; import com.google.common.collect.AbstractIterator; import com.google.common.io.CharSource; import com.google.common.io.Closeables; import com.palantir.spark.binarystream.data.PortableFile; import com.palantir.transforms.lang.java.api.Compute; import com.palantir.transforms.lang.java.api.FoundryInput; import com.palantir.transforms.lang.java.api.FoundryOutput; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import java.io.Reader; import java.nio.charset.StandardCharsets; import java.util.Iterator; import java.util.zip.GZIPInputStream; import java.util.zip.ZipInputStream; import org.apache.spark.TaskContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; import org.apache.spark.util.TaskCompletionListener; /** * この例では、.csv/.gz/.zip ファイルを入力として想定しています。 * <p> * Spark 上で次の操作を並列に実行します。 * <ol> * <li>ファイルの形式を検出します。</li> * <li>ファイル形式が .gz または .zip の場合、ファイルを解凍します。</li> * <li>解凍した .csv ファイルのスキーマを推論します。</li> * <li>.csv の行のデータセットを、そのスキーマのデータセットに変換します。</li> * </ol> */ public final class UnzipWithSparkToDataset { @Compute public void compute(FoundryInput input, FoundryOutput output) { // 入力ファイルの Spark データセットを取得します。 Dataset<PortableFile> files = input.asFiles().getFileSystem().filesAsDataset(); // 入力ファイルのデータセットを .csv の行のデータセットに変換します。 Dataset<String> csvDataset = files.flatMap((FlatMapFunction<PortableFile, String>) portableFile -> // 現在の入力ファイルから InputStream を取得します。 portableFile.convertToIterator(inputStream -> { String fileName = portableFile.getLogicalPath().getFileName().toString(); // .gz および .zip ファイルを検出し、各ファイルから行イテレーターを取得します。 if (fileName.endsWith(".gz")) { return new InputStreamCharSource(new GZIPInputStream(inputStream)).getLineIterator(); } else if (fileName.endsWith(".zip")) { return new ZipIterator(new ZipInputStream(inputStream)); } else { return new InputStreamCharSource(inputStream).getLineIterator(); } }), Encoders.STRING()); // スキーマを推論し、.csv の行のデータセットをそのスキーマのデータセットに変換します。 Dataset<Row> dataset = files .sparkSession() .read() .option("inferSchema", "true") .csv(csvDataset); output.getDataFrameWriter(dataset).write(); } /* * この ZipIterator は、アーカイブ内のすべてのファイルが同じスキーマの .csv ファイルであり、 * 同じデータセットに属することを前提としています。 */ private static final class ZipIterator extends AbstractIterator<String> { private Iterator<String> lineIterator; private ZipInputStream zis; ZipIterator(ZipInputStream zis) throws IOException { this.zis = zis; lineIterator = new InputStreamCharSource(zis).getLineIterator(); } @Override protected String computeNext() { if (!lineIterator.hasNext()) { // 行イテレーターに次の要素がない場合、次のファイルがあるかどうかを確認します。 try { // 次の空でないファイルを探します。 while (zis.getNextEntry() != null) { lineIterator = new InputStreamCharSource(zis).getLineIterator(); if (lineIterator.hasNext()) { break; } } return lineIterator.hasNext() ? lineIterator.next() : endOfData(); } catch (IOException e) { throw new RuntimeException(e); } } else { return lineIterator.next(); } } } private static final class InputStreamCharSource extends CharSource { private final Reader inputStream; private InputStreamCharSource(InputStream inputStream) { this.inputStream = new InputStreamReader(inputStream, StandardCharsets.UTF_8); } @Override public Reader openStream() throws IOException { return inputStream; } @SuppressWarnings("MustBeClosedChecker") Iterator<String> getLineIterator() { try { return super.lines().iterator(); } catch (IOException e) { throw new RuntimeException(e); } finally { if (TaskContext.get() != null) { // Spark で実行している場合、タスクが完了したときにストリームを閉じます。 TaskContext.get().addTaskCompletionListener((TaskCompletionListener) context -> close()); } else { close(); } } } private void close() { Closeables.closeQuietly(inputStream); } } }
この例では、.zip ファイルを入力として受け取り、すべての入力データセットファイルを単一の .zip ファイルに結合します。このトランスフォームでの計算は並列化されません。
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48/* * (c) Copyright 2018 Palantir Technologies Inc. すべての権利を留保します。 */ package com.palantir.transforms.java.examples; import com.google.common.io.ByteStreams; import com.palantir.transforms.lang.java.api.Compute; import com.palantir.transforms.lang.java.api.FoundryInput; import com.palantir.transforms.lang.java.api.FoundryOutput; import com.palantir.transforms.lang.java.api.ReadOnlyLogicalFileSystem; import com.palantir.transforms.lang.java.api.WriteOnlyLogicalFileSystem; import com.palantir.util.syntacticpath.Paths; import java.util.zip.ZipEntry; import java.util.zip.ZipOutputStream; /** * データセット内のすべてのファイルを1個の大きな .zip ファイルにまとめる例です。 * <p> * 並列に実行することが難しいため、処理は単一のスレッドで実行されます。 * <p> * 警告：一般に、Spark を活用するには {@link UnzipWithSpark} および {@link UnzipWithSparkToDataset} の API を * 使用することが推奨されます。これは効果的な並列化が難しい計算の例であり、 * そのため、ドライバー上でのみファイルシステム操作を使って実行しています。 */ public final class ZipOnDriver { @Compute public void compute(FoundryInput zipFiles, FoundryOutput output) { ReadOnlyLogicalFileSystem inputFileSystem = zipFiles.asFiles().getFileSystem(); WriteOnlyLogicalFileSystem outputFileSystem = output.getFileSystem(); // 出力データセットのファイルシステム内の「bigzip.zip」というファイルに書き込みます。 outputFileSystem.writeTo(Paths.get("bigzip.zip"), outputStream -> { // 各ファイルを同じ .zip ファイルに書き込むため、OutputStream を ZipOutputStream でラップします。 try (ZipOutputStream zipOutputStream = new ZipOutputStream(outputStream)) { // 入力データセットの FileSystem 内の各ファイルを読み取り、 // 「bigzip.zip」に新しいエントリーを指定してから、バイトをコピーします。 inputFileSystem.listAllFiles().forEach(inputPath -> { inputFileSystem.readFrom(inputPath, inputStream -> { zipOutputStream.putNextEntry(new ZipEntry(inputPath.toString())); ByteStreams.copy(inputStream, zipOutputStream); return null; }); }); } }); } }