トランスフォームに関するよくある質問

トランスフォームに関するよくある質問を以下に示します。

一般的な情報については、トランスフォームのドキュメントを参照してください。

• transforms-python で Parquet の代わりに CSV ファイルを保存できますか？
• 1個の Python トランスフォームから複数の出力データセットをビルドできますか？
• トランスフォームで GZIP ファイルを開くにはどうすればよいですか？
• Foundry パイプラインの一部としてファイルを解凍するにはどうすればよいですか？並列で解凍できますか？

---

transforms-python で Parquet の代わりに CSV ファイルを保存できますか？

各トランスフォーム言語でこれを行う例を以下に示します。

Java

Copied!
1
2
3
foundryOutput.getDataFrameWriter(dataFrame)
.setFormatSettings(DatasetFormatSettings.builder().format("csv").build())
.write();

Python

Copied!
1
2
3
4
5
6
7
from transforms.api import transform, Input, Output
@transform(
  output=Output("/path/to/python_csv"),
  my_input=Input("/path/to/input")
)
def my_compute_function(output, my_input):
  output.write_dataframe(my_input.dataframe(), output_format="csv")

SQL

Copied!
1
CREATE TABLE `/path/to/sql_csv` USING CSV AS SELECT * FROM `/path/to/input`

ページの先頭に戻る

---

1個の Python トランスフォームから複数の出力データセットをビルドできますか？

複数のトランスフォームやデータセットが必要な場合は、for ループを使用して作成できます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from transforms.api import transforms_df, Input, Output

def transform_generator(sources):
	#type: (List[str]) -> List([transforms.api.Transform])
	transforms = []
		# この例では複数の入力データセットを使用します。また、複数の出力を
		# 1個の入力データセットから生成することもできます。
	for source in sources:
		@transforms_df(
			Output('/sources/{source}/output'.format(source=source)),
			my_input=Input('/sources/{source}/input'.format(source=source))
			)
		def compute_function(my_input, source=source):
			# 関数内で source 変数をキャプチャするには、デフォルト値付きのキーワード引数として渡します。
			return my_input.filter(my_input.source == source)

		transforms.append(compute_function)

	return transforms

TRANSFORMS = transforms_generator(['src1', 'src2', 'src3'])

これで、モジュールの TRANSFORMS 属性をインポートし、各トランスフォームを手動でパイプラインに追加できます。

Copied!
1
2
3
4
import my_module

my_pipeline = Pipeline()
my_pipeline.add_transforms(*my_module.TRANSFORMS)

1個の入力を受け取り、同じビルドで複数のデータセットを出力する単一のトランスフォームにするには、以下のようにプログラムで実現することもできます。

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
# `/examples/students_hair_eye_color` データセットを使用します
students_input = foundry.input('/examples/students_hair_eye_color')
students_input.dataframe().sort('id').show(n=3)
+---+-----+-----+----+
| id| hair|  eye| sex|
+---+-----+-----+----+
|  1|Black|Brown|Male|
|  2|Brown|Brown|Male|
|  3|  Red|Brown|Male|
+---+-----+-----+----+
Note that this example only shows the top three rows.

from transforms.api import transform, Input, Output

@transform(
	hair_eye_color=Input('/examples/students_hair_eye_color'),
	males=Output('/examples/hair_eye_color_males'),
	females=Output('examples/hair_eye_color_females'),
)
def brown_hair_by_sex(hair_eye_color, males, females):
	# type: (TransformInput, TransformOutput, TransformOutput) -> None
	brown_hair_df = hair_eye_color.dataframe().filter(hair_eye_color.dataframe().hair == 'Brown')

	males.write_dataframe(brown_hair_df.filter(brown_hair_df.sex == 'Male'))
	females.write_dataframe(brown_hair_df.filter(brown_hair_df.sex == 'Female'))

トランスフォームに関する詳しいヘルプや情報については、以下のドキュメントを参照してください。

• Python トランスフォーム
• Java トランスフォーム
• Spark SQL トランスフォーム

ページの先頭に戻る

---

トランスフォームで GZIP ファイルを開くにはどうすればよいですか？

トランスフォームへの入力はストリームを基盤とするファイルのようなオブジェクトなので、ファイルとして処理できます。そのため、ファイル全体をメモリーに読み込んだり、ディスクにコピーしたりすることを気にする必要がなく、はるかに大きなファイルを使用できます。

Python 3 に含まれる gzip パッケージと io パッケージを使用します。

Copied!
1
2
3
4
5
6
7
import gzip, io

def process_file(file_stauts):
	fs = input_dataset.filesystem()
	with fs.open(file_status.path, 'rb') as f:
		gz = gzip.GzipFile(fileobj=f)
		br = io.BufferedReader(gz)

また、読み取り時に文字列を返すようにする場合は、ラップできます。

Copied!
1
tw = io.TextIOWrapper(br)

ファイルにエンコーディングがある場合は、それを指定できます。

Copied!
1
tw = io.TextIOWrapper(br, encoding='CP500')

トランスフォームに関する詳しいヘルプや情報については、以下のドキュメントを参照してください。

• Python トランスフォーム
• Java トランスフォーム
• Spark SQL トランスフォーム

ページの先頭に戻る

---

Foundry パイプラインの一部としてファイルを解凍するにはどうすればよいですか？並列で解凍できますか？

以下の例では、Java と Spark を使用して、zip アーカイブ内の各ファイルを並列で解凍します。単一の圧縮ファイル内で解凍処理を並列化する場合は、.bz2 のような分割可能なファイル形式を使用します。

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

トランスフォームに関する詳しいヘルプや情報については、以下のドキュメントを参照してください。

• Python トランスフォーム
• Java トランスフォーム
• Spark SQL トランスフォーム

ページの先頭に戻る