apache-spark

1熱

1答えて

は、変換の次のタイプを行うための推奨方法はあります：入力： userID timestamp action u1 100 edit1 u1 122 edit2 u1 135 edit3 u2 143 edit4 u2 158 edit5 u1 212 edit6 u1 241 edit7 出力：失われたセッション情報でユーザーIDの結果によってグルー

2熱

1答えて

スパークをひどく誤ってcsvを誤って修正する方法はありますか？

私は2つのcsvファイルを持っています.1つは映画のためのキーワードを含み、もう1つはキャストとクルーを含みます。 keywords.csvファイルには、次のようになります。 $ head -n 3 keywords.csv id,keywords 862,"[{'id': 931, 'name': 'jealousy'}, {'id': 4290, 'name': 'toy'}, {'id'

1熱

1答えて

groupByKeyで値を追加

scalaとSparkのgroupByKeyにいくつかの問題があります。私は2ケースクラスがあります。 case class Employee(id_employee: Long, name_emp: String, salary: String) 私は、この第二ケースクラスを使用した瞬間のために： case class Company(id_company: Long, employee:

1熱

2答えて

マルチスレッドドライバからApache Spark SQLジョブを起動する

Sparkを使用して約1500のリモートOracleテーブルからデータを取得したいと思っていましたが、スレッドごとにテーブルをピックアップするマルチスレッドアプリケーションそれぞれのテーブルから読み取るスパークジョブを起動します。スパークは、アプリケーション間でのスケジューリングのための設備を提供で実行されていることを公式火花サイトからそれは、これが働くことができることは明らかだhttps://

0熱

1答えて

PySpark：パーティション分割せずに出力ファイルを別々のファイルに書き込む

非常に大きなgzipファイルがあります。私はPySparkを使用して、後で使用するためにParquet形式のS3にあるすべてのファイルを再保存しようとしています。単一ファイルの場合は（例では、2012年6月1日）は私が行います dataframe = spark.read.csv('s3://mybucket/input/20120601.gz', schema=my_schema, heade

-2熱

1答えて

どのように私はPythonクラスからオブジェクトを作成し、sparkのAPIの下で書かれたJavaプログラムで使用することができます

私はJavaで書かれたプログラムを持っています。このプログラムは、データベースからデータを読み取り、各レコードを分析し、その結果を再度データベースに書き込む。 JavaでスパークのAPIでコードを書き直そうとしていますが、各レコードの解析の一部はPythonコードで行われています。私の質問は、どのようにJava SparkのAPIの下で書かれたJavaコードで私のPythonオブジェクトを呼び出す

0熱

1答えて

カスタマイズされたパーティショナーがsparkでローカル変数を持たない

spark用にカスタマイズされたパーティショナー機能をPythonで作成するのを手伝ってください。私はエントリのデータキーとパーティションIDの間のマッピングを伝えるファイルを持って、私が最初にスパーク sc.parallelize(input_lines).partitionBy(numPartitions=xx, partitionFunc=lambda x : data_to_partit

1熱

1答えて

pyspark csvをデータフレームに書き込むことなくディスクに書き込む

URLにあるcsvをディスクに書き込まずにPysparkのデータフレームに読み込むにはどうすればよいですか？ノー運と、次の試してみた： import urllib.request from io import StringIO url = "https://raw.githubusercontent.com/pandas-dev/pandas/master/pandas/tests/dat

1熱

3答えて

hbaseの行のデータセットを保存する

私はKafkaからデータ（json文字列）を読み込み、それを解析して与えられたスキーマを持つデータセットに変換し、そのデータセットの属性に対する集計を実行します。私はこのデータセットをjson文字列に変換してからhbaseに格納したいと思っています。これを行う最善の方法は何ですか？サンプルデータセット：HBaseの中に私のキーに対する id|name|age 1 |geet|21 予想さ

1熱

1答えて

同じIDが常に同じロジックプランで出力されますか？

あなたの下には、私がやろうとしているものの簡略版があります。 S3に格納されているを150 parquets（> 10TB）にロードします。次に、このデータフレームにfunc.monotonically_increasing_id()というID列を与えます。その後、私はこのdataframeの2つの偏差を保存します。私が適用機能は、今私はparquet_2を作成するために、再びS3からすべてのデー