scalding

6熱

1答えて

スケーリング：groupBy（ 'field）{。size}の後に他のフィールドを保持する方法？

だから私の入力データは、二つのフィールド/列があります。ID1 & ID2を、そして私のコードは次のようである：（私は仮定するもの）で TextLine(args("input")) .read .mapTo('line->('id1,'id2)) {line: String => val fields = line.split("\t") (fields(0),fiel

0熱

2答えて

ScaldingでList [String]のdiffを生成する

私の最初の値がidで、2番目がstuffであるスケーリングジョブにrecords:TypedType[(String, util.List[String])]があります。次想像：私は出力に指定したidのために互いに異なるレコードのみを希望 ("1", ["a","b","c"]) ("1", ["a","b","c"]) ("1", ["a","b","c"]) ("2", ["a","b"

0熱

1答えて

スケーリング時の垂直パーティショニング

私はTypedTipe[(String, String, Long)]を持っていますが、最初のStringは限られた数（〜10）しか取れません。私の出力を分割して、各タイプ（最初の文字列の名前を持つI.E. 10フォルダ）ごとにフォルダを作成するようにしたいと思います。これはHiveで実現するのは簡単ですが、私はScaldingでそれを行うエレガントな方法を見つけることができません。メソッドdef

7熱

2答えて

Scalding mapメソッドでタプルではなくコレクションを出力できますか？

Scaldingで小さいフィールドから22フィールド以上のパイプを作成する場合は、Scalaタプルによって制限されます。スカラタプルは22個を超えることはできません。タプルの代わりにコレクションを使用する方法はありますか？私は悲しいことに動作しない次の例のようなものを想像します。 input.read.mapTo('line -> aLotOfFields) { line: String =>

0熱

1答えて

Scaldingでキーのシリアル化方法を変更するにはどうすればよいですか？

私は火傷の仕事にカスタムタイプによってグループ分けしています： typedPipe .map(someMapper) .groupBy(_.nonPrimitiveField) .sum .write(sink) を私の出力では、キーが有効ではありませんtoString出力として現れます。どのように私はこれらのキーのためのカスタムシリアライザを使用するこ

0熱

1答えて

TypedType [[MyObject]]をTypedType [MyObject]に設定する

私のスケーリングフローにはTypedType[Set[MyObject]]があります。それをTypedType[MyObject]にフラット化するにはどうすればよいですか？

0熱

1答えて

検索最大[文字列、セット[のMyStuff]

私は次の操作を行う必要がありますいくつかのStringプロパティによって、私の記録が取得大きすぎる廃棄グループグループをここで残りのグループのそれぞれで最大の要素は、私がこれまでに得たものである： val records: TypedPipe[MyStuff] = ... records .groupBy(_.getThatStringProperty) .toSet

5熱

4答えて

倍

の値でグループを使用してGROUPBY foldLeftを（火傷）のようなデータがあります： pid recom-pid 1 1 1 2 1 3 2 1 2 2 2 4 2 5 はそれを行う必要があります。 pid, recommendations 1 2,3 2 1,4,5 意味は2列目からの自己を無視し、そして作りますコンマ区切りの文字列にしてください。そのタブ区切り

0熱

1答えて

Hortonwork HDP 2.1はどのバージョンのScalaをサポートしていますか？

Scalding 2.10.4を使用してスケーリングジョブをビルドしています。その仕事を成功させる。しかし、私がHortonworksで仕事をすると、次の例外がスローされます。いくつかの回答によると、この問題は、競合Scalaのバージョンが原因発生します。この問題を教えていただけますか？私はそれを見つけることができない私のホートンワークスでスカラー版を探した。私は間違って何かしていますか？

0熱

1答えて

スケーリングジョブを入力バケット上で再帰的に動作させるにはどうしたらいいですか？

私はEMRで動作するスケーリングジョブを持っています。これは、複数のファイルを含むS3バケット上で実行されます。私はそれがネストされたバケツ、自身がファイルが含まれているバケットを含む、すなわちバケット上で実行したい MultipleTextLineFiles("s3://path/to/input/").read /* ... some data processing ... */