pyspark-sql

0熱

1答えて

これは悪い考えかもしれないので、pyspark/bigdataに少し新しかったですが、私は約100万個の個別のCSVファイルを持っていて、それぞれにいくつかのメタデータが関連付けられています。私はすべてのメタデータフィールドの列を持つpysparkデータフレームを望みますが、エントリがメタデータの各セットに関連付けられた（全体の）CSVファイルである列も必要です。私は仕事中ではありませんが、ほぼ

1熱

1答えて

PySpark Dataframes：複数の条件をコンパクトコードでフィルタリングする方法は？

カラム名のリストがあり、それらのカラムの値がゼロより大きい場合にローをフィルタリングしたい場合は、これに似たものがありますか？ columns = ['colA','colB','colC','colD','colE','colF'] new_df = df.filter(any([df[c]>0 for c in columns])) これが返されます。 ValueError: Canno

-2熱

1答えて

Spark DataFrameとWindow関数を使用してローリング平均リターンを計算する

sparkデータフレームで初めての作業です。ウィンドウ関数を使用して毎日の各ストックの平均1日のリターンを計算する方法を理解しようとしています。ティッカーでグループ化しようとしていて、ローリング差分ウィンドウ関数を適用しようとしていますが、ウィンドウ関数やその動作方法について多くのドキュメントが見つかりません。私が持っているデータは、日付、オープンプライス、高価格、低価格、終値、取引量、および

1熱

1答えて

1列ごとにDataFrameを変換してpysparkに2列の新しい列を作成するにはどうすればいいですか？

私は、データが、そのステータスと値に基づいてフィルタリングされている形式にこのデータフレームを変換する二つの列「X1」と「X2」は彼らのしているデータフレームの「x」、 x1(status) x2 kv,true 45 bm,true 65 mp,true 75 kv,null 450 bm,null 550 mp,null 650 を持っています x1 true nu

7熱

1答えて

Spark 2.0：GetOrCreateでSparkSessionパラメータを再定義し、WebUIで変更を表示しない

私はpySparkでSpark 2.0を使用しています。私は2.0で導入されたGetOrCreate方法を通じてSparkSessionパラメータを再定義しています。そこに有効なグローバルなデフォルトSparkSessionであり、そうならば、その1を返すかどうかをこのメソッドに最初のチェックを。有効なグローバルデフォルトSparkSessionが存在しない場合、このメソッドは新しいSpark

0熱

1答えて

Pygark JDBCコネクタを使用してPostgresでPostgres SQL関数をリモートで実行するには？

私はJDBC connectorを使用してPySparkアプリケーションからリモートのPostgresサーバ上で次のクエリを実行したい： SELECT id, postgres_function(some_column) FROM my_database GROUP BY id 問題は、私は明らかにpostgres_functionので、spark.sql(QUERY)を使用してPyspark

4熱

1答えて

EARツールを交換することができます

既存のプロセス - 生の構造データがRedshiftのステージングレイヤーにコピーされます。その後、Informatica、TelendなどのETLツールを使用して、Datamart/datawarehouseのFactおよびDimensionテーブルにインクリメンタルロードを行います。すべての結合はデータベースレイヤー内で行われます（ETLはクエリをDBにプッシュします） - SparkはETL

3熱

1答えて

スパークリクエスト最大数

私はスパークの初心者です。私は最も訪問されたウェブページを取得できるようにリクエストしています。私の要求は、私は最大のカウントを有する唯一のデータフレーム取得この要求に以下の mostPopularWebPageDF = logDF.groupBy("webPage").agg(functions.count("webPage").alias("cntWebPage")).agg(functio