pyspark-sql

1熱

1答えて

複数のJSONオブジェクトを1行のファイルでPySparkに読み込むときのヌル値

S3バケットからPySpark DataFrameに一連のJSONファイルを取得する際に助けが必要です。このバケットのファイルはすべて、拡張子が.jsonですが、残念ながら通常のSparkでは1行に1つのJSONオブジェクトが必要ですが、大括弧で囲まれた1行になります。ので、代わりの： {"first_column": 12, "second_column": {"nested_column"

0熱

1答えて

3つのDataFrameを1つのDataFrameに結合するにはどうすればよいですか？

私は1つの中で一緒に三つの異なるデータフレームに参加しようとしていますが、私はすべての3つに参加することに問題を抱えています。私は2つに参加することができました。にはどうすれば正しく3つのデータフレームに参加することができますか？ Pyspark：以下 1.6.0は私がこれまでに取り組んでいる： # EXPECTED OUTPUT: # -------file1.csv---------|-

0熱

1答えて

私は何をしていてもメモリを消耗します

大きなデータフレームをフィルタリングするためにsparkを使用しようとしています。パンダのデータフレームとしては、約70GBのメモリがあります。私はディスクにスワップする必要があるので、私はしかし、本当にゆっくりと、パンダを使用してこのデータをロードし、フィルタリングすることができる午前などしかし、私は、私はJavaからメモリエラーに遭遇PySparkを使用してこれをやってみたとき。 from

0熱

1答えて

スパークSQLクエリは、ハードコードされた値で動作しますが、可変

で私はこのようなSQLクエリがあるとしたPySparkプログラムを持っていない： id = 'abc1' info = sqlc.sql("SELECT * from df WHERE isn = 'id'") このクエリは動作しません。私は値の空白のセットで返されます。しかし、私が値をハードコードするならば、それは動作します。 info = sqlc.sql("SELECT * from

-1熱

1答えて

Pysparkの新しいDataFrameに繰り返し保存する

私は3つのPySpark DataFramesに基づいて計算を行っています。このスクリプトは、計算を実行するという意味では機能しますが、私は計算の結果を正しく処理するのに苦労しています。列のそれぞれについて、およびデータの行ごとに import sys import numpy as np from pyspark import SparkConf, SparkContext, SQLCon

3熱

2答えて

pysparkのリスト内の異なるデータフレーム列を合計する正しい方法は何ですか？

スパークデータフレーム内の異なる列を合計します。コード＃2に近づくされていないのはなぜ from pyspark.sql import functions as F cols = ["A.p1","B.p1"] df = spark.createDataFrame([[1,2],[4,89],[12,60]],schema=cols) # 1. Works df = df.withCo

0熱

2答えて

pysparkから辞書を作成する最速の方法DF

私はsqlppydataとpysparkを使ってSQLクエリを実行し、出力DFを辞書に変換してmongoに一括挿入しています。私はスパークDFの辞書への変換をテストするために、多くの同様の質問を行ってきました。現在、私はこのメソッドをbulk DFに変換するためにこのメソッドを使用しています。そして、10Kレコードで2〜3秒かかります。私は私のアイデアをimpliment方法を以下に記載しま

0熱

1答えて

PySpark Pythonで2つの日付列を引く

私はPySpark DataframeのPythonで2つの列を減算しようとしています。私はそれをやっている問題がたくさんあります。私はタイムスタンプとして列の型を持っていて、列はdate1 = 2011-01-03 13:25:59で、コラムdate2 = 2011-01-03 13:27:00はので、私はdate2 - date1したいと私は PySaprkでこれを行うことができますどのような

0熱

1答えて

PySpark 2.2.0 S3へのDataFrameの書き込みAmazonServiceExceptionクラスが見つかりません

私はpysparkでS3にSpark DataFrameを書き込もうとしています。私はSparkバージョン2.2.0を使用しています。 sc = SparkContext('local', 'Test') sc._jsc.hadoopConfiguration().set("fs.s3a.awsAccessKeyId", aws_key) sc._jsc.hadoopConfiguration