Questions tagged [pyspark]
The Spark Python API (PySpark) exposes the apache-spark programming model to Python.
9,532
questions with no upvoted or accepted answers
0
votes
0answers
51 views
find the sum on json list based on condition in pyspark
Nested Json sample:
{
"accounts": {
"1": [{
"userId": "XXXXXXX",
"accountId": "1",
"...
0
votes
0answers
23 views
can't run example, error '-Xmx1g org.apache.spark.deploy.SparkSubmit'
i'm trying to replicate wordcount example from cloudera website
spark-submit --master yarn --deploy-mode client --conf
"spark.dynamicAllocation.enabled=false" --jars
SPARK_HOME/examples/...
0
votes
1answer
260 views
Percentile rank in pyspark using QuantileDiscretizer
I am wondering if it's possible to obtain the result of percentile_rank using the QuantileDiscretizer transformer in pyspark.
The purpose is that I am trying to avoid computing the percent_rank over ...
0
votes
0answers
46 views
How to make a recommendation system based on implicit data on spark?
I have a dataset of purchased items like this:
+-------+-----------+--------------+--------------+
|user_id| item_id| category | tag |
+-------+-----------+--------------+--------------...
0
votes
1answer
22 views
Wring files in s3 using spark and reading the same using pandas dataframe
writing a file s3 using spark usually creates a directory with two files success and the other file name starts with name as part which has actual data in s3 , how to load the same file using pandas ...
0
votes
0answers
19 views
Significance of raise in python logger
am new to python
Actually am trying to display logging.error message in runtime console itself during execution ( Pyspark)
Code 1 :
try:
if( len(sys.argv) ==3 ):
log.info("print message &...
0
votes
1answer
322 views
Spark Event log directory
I am using PySpark (standalone without hadoop etc) and calling my pyspark jobs below and it works fine:
PYSPARK_PYTHON=python3 JAVA_HOME="/usr/lib/jvm/java-8-openjdk-amd64/jre" SPARK_HOME=~/....
0
votes
0answers
30 views
Spark exception 5063 in TensorFlow extended example code on GPU
I am trying to run an example code of TensorFlow extended at https://www.tensorflow.org/tfx/tutorials/transform/census
on databricks GPU cluster.
My env:
7.1 ML Spark 3.0.0 Scala 2.12 GPU
python 3.7
...
0
votes
1answer
33 views
Row delimiter change while writing a tsv file in adls gen 1 from databrick
I want to save the tsv file to adls gen1. Using the below command to save the data but it writing a row delimiter as "\n"(LF) I want to writing a row delimiter "\r\n"
df.coalesce(...
0
votes
2answers
97 views
PySpark SQL overwriting returns empty table
I'm migrating some data within a table, I am trying to change the value of the 'date' column, but it seems like PySpark erases the data while it is reading it.
I am doing the following steps:
read ...
0
votes
1answer
89 views
How to read data from Databricks DBFS using Rest API in csv or Excel format?
I am using Databricks Rest API to read datasets stored on DFBS. The output is coming in 64bit encoded format and in json format.
I need the output in tabular format which is easy to read.
Output of ...
0
votes
0answers
31 views
Converting Parquet to csv and vice versa without schema issues by using Pyspark dataframe
Converting Parquet to csv and vice versa without schema issues by using Pyspark dataframe.
I did the following:
df=spark.read.parquet("part-00000-caae021e-9d02-46e8-a38d-5eaee04f1a03-c000.snappy....
0
votes
0answers
132 views
An error occurred while calling o266.fit while trying to fit the train data set in sparknlp pipeline
An error occurred while calling o266.fit while trying to fit the train data set in sparknlp pipeline.
Environment:
openjdk version "1.8.0_252"
Spark NLP version 2.5.0
Apache Spark version: 2....
0
votes
2answers
270 views
combine and count a json column in spark dataframe
I would like to aggregate a column values (json) in spark dataframe and hive table.
e.g.
year, month, val (json)
2010 01 [{"a_id":"caes"},{"a_id":"...
0
votes
1answer
426 views
Getting the 'Exception thrown in awaitResult:' error when trying to copy table in glue to redshift
I have been trying to copy a table over from glue to one in redshift. I created a job with the following code
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
...
0
votes
0answers
36 views
Spark job runs good with only one executor but same job fails when running with two executors
Error Message:
ERROR storage.ShuffleBlockFetcherIterator: Failed to get block(s) from WN2
java.io.IOException: Failed to connect to WN2
Caused by: io.netty.channel.AbstractChannel$...
0
votes
0answers
26 views
TaskCompletionListenerException: Data read has a different length than the expected
While writing a parquet file to S3, I keep encountering the following issue
org.apache.spark.util.TaskCompletionListenerException: Data read has a different length than the expected: dataLength=...
0
votes
0answers
49 views
Pyspark/DataProc: How to Write Chunks To Temporary File?
I need to read in a Text/CSV response from an API (~40 GB), which I am attempting to chunk and append each chunk to a csv file. Then I will have Pyspark read in the file as a spark dataframe. ...
0
votes
0answers
25 views
Pyspark function to affect values with condition
I want to structurate my dataset with rules I made but I can't find how to do even if the problem doesn't seem very complicated
I have a dataset structurate like this :
Id type latitude ...
0
votes
1answer
64 views
Standardize all timestamps with multiple timezones in a Spark DataFrame to UTC format
I am using pyspark with the purpose of standardizing a column of type string that contains timestamps in different formats, and different time zones, for instance:
| Timestamp |
|------...
0
votes
0answers
96 views
Kafka with Spark Streaming works in local but it doesn't work in Standalone mode
I'm trying to use Spark Streaming with a very simple script like this:
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import ...
0
votes
0answers
16 views
How to select survival times for right-censored data?
I am fitting an AFT survival model in PySpark, and I have a question related to survival times for censored data. I am studying the purchase behaviour of customers. The dependent variable of my model ...
0
votes
0answers
35 views
Enable to find executors in TF Cluster while using TensorFlowOnSpark
I am testing a spark deep learning library TensorFlowOnSpark using the code specified in their Github repo.
https://github.com/yahoo/TensorFlowOnSpark/blob/master/examples/mnist/keras/mnist_spark.py
...
0
votes
0answers
44 views
What is the meaning of the inputs in spark Power Iteration Clustering Algorithm
In Spark Documentation, it has described the parameters for power iteration clustering algorithm as below
k: the number of clusters to create
initMode: param for the initialization algorithm
maxIter: ...
0
votes
0answers
156 views
pypark , pandas udf, pyarrow.lib.ArrowInvalid error can't convert 781127.0 to double
I have a simple pyspark pandas udf function.
With short period it works OK, and for longer period of time. It complains that can't convert 781127.0 to double. The number comes from MARKET_VALUE_LOCAL, ...
0
votes
0answers
21 views
Accessing Hive in AWS EMR from a remote web application
I currently building a Flask application that uploads files to an S3 bucket, but I also need it to be able to use PySpark/Python to create tables in Hive (in EMR). It's easily done running a pyspark ...
0
votes
0answers
36 views
How to do rollup in spark SQL
I have following query
dimSQL = '''select * from (
select dimension1, dimension2, dimension3, dimension4,
grouping(dimension1)c1, grouping(dimension2)c2, grouping(...
0
votes
0answers
27 views
How can I better optimize this Hive query for speed?
What I am looking to do is better optimize these Hive queries in PySpark with Spark Context, as seen below, for better speed. Right now, when doing the Hive table comparison for big tables, it ends up ...
0
votes
0answers
16 views
Import Pyspark Script to Python Script
I am trying to import PySpark script into Python Script and execute, but it gives error saying no module found.
Are there any specifics i am missing while importing pyspark script?
0
votes
1answer
28 views
PySpark : Performance optimzation during create_map for categorizing the page visit
I am working on optimizing the below operation whose exection time is relatively high on the actual dataset(large datset).I tried below on two of the pyspark dataset 1 & 2 to arrive at the "...
0
votes
0answers
7 views
Spark createstream works with Zookeeper in local but it doesn't work with Zookeeper in remote
I'm trying to use Spark Streaming with Kafka. When I have all the components in one machine it works well. But if I try to put Zookeeper into another machine, it doesn't work. If I try to use the ...
0
votes
0answers
45 views
NULLs when loading data with a schema in Pyspark
I would like to load data with Pyspark (from AWS s3) using a custom schema for the data types (as I did below). If I do this though I get NULL for all rows and columns. If I comment out .schema(...
0
votes
0answers
129 views
Error while EMRFS Consistency View enabled along with Hudi
Environment: EMRFS Consistency enabled in EMR cluster (5.29.0) & KMS is enabled for both Dynamodb and s3
We are exploring Hudi for insert & upsert operations. While inserting data into Hudi ...
0
votes
0answers
75 views
Perform data upsert in postgres via Pyspark
I want to perform an Upsert for approximately 2.5 Million records on a large 'Table A' in postgres. For which I am trying to push the data to a temporary table 'Table B' via jdbc connector using ...
0
votes
1answer
139 views
Versions of Python & Spark to work with VS Code Notebooks
I'm developing scripts for AWS Glue, and trying to mimic development environment as close as possible to their specs here. Since it's a bit costly to run a Notebook server/development endpoint, I set ...
0
votes
1answer
150 views
pandas df to spark df conversion takes long time on Databricks notebook
I have a pandas df which is over 10 million in rows. I'm trying to convert this pandas df to spark df using the below method.
spark_session = SparkSession.builder.appName('pandasToSparkDF')....
0
votes
0answers
36 views
Exception on PySpark when write to Parquet
I am working with huge data at JSON format that is coverted to parquet using pySpark. In gral, all it is ok, but currently I am having issue on some transformation. But I could not debug or I don;t ...
0
votes
0answers
24 views
removing repeated columns in pivot pyspark dataframe
I have below parquet file
Time source event field_1 field_2 field_3 .......com_1 com_2...com_n
10:00 device1 ev1 10 start 1.5 .03
10:00 device1 ev1 ...
0
votes
0answers
15 views
Lots of yarn resources consumed after making .toPandas()
I’m using spark 2.2, i have a spark dataframe that contains about 3millions of rows and 15 columns. When i’m applying .toPandas() on my dataframe, lots of yarn resource is consumed, about 87GB ...
0
votes
0answers
46 views
pyspark + keras as pandas_udf does not work properly
In my scenario I wish to train a keras model per each partition on a Spark cluster with Python enabled.
For this I am using Spark PandasUDFs.
Inside the PandasUDF code I have the following command:
...
0
votes
0answers
99 views
An error occurred while calling o337.load. : java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/writer/StreamWriteSupport
I am new to spark structured streaming. I am writing below code in EMR notebook.
We need extra jar for spark structured streaming to read kafka stream which i downloaded from
https://mvnrepository.com/...
0
votes
0answers
27 views
how to send data to pyspark form inside python file?
I have scraped the data from a website, it is a list of lists:
data = [
['name', '20 July 1995 08:05', 0, 'text text text'],
['name', '20 July 1995 08:05', 0, 'text text text'],
['name', '20 July 1995 ...
0
votes
1answer
167 views
S3 compaction by identifying the smaller-multiple parquet files of a dataframe into single partition dataframe by using pyspark
I have requirement to read the multiple-smaller parquet files of a dataframe(s3-folder) and re-write them into same location with single/more files of size range in b/w min:128mb and Max:900mb.
...
0
votes
0answers
16 views
Spark in K8 - WARN TaskSchedulerImpl: Initial job has not accepted any resources. But plenty of resources actually
I have a spark running in a K8 environment: one master and three workers.
I have the following code in python 3.7.5 running from my machine:
import pyspark
from pyspark import SparkConf, SparkContext
...
0
votes
0answers
89 views
Building Custom Stringindexer in Pyspark
I would like to build my own StringIndexer in PySpark where the indexing is based on a weight column. I know how to code up the indexing but have no idea how to turn this in an estimator (that also ...
0
votes
0answers
49 views
Spark model fit throwing java.lang.OutOfMemoryError: Java heap space
i have a python code for training and testing machine learning models using pyspark, the code is below:
def build_model(self, database_url_training, database_url_test, label):
training_file = self....
0
votes
2answers
82 views
How to filter a nested column of type string in a dataframe using PySpark
I have a nested column of type string in a dataframe and I am trying to filter on it based on non null values
install_activity_touch_data_additional_data_custom_fields: string (nullable = true) and I ...
0
votes
0answers
17 views
Pyspark load file and specify data type
I would like to load a csv file with PySpark (from AWS S3). I have one column which is specified as integer after loading the data but i would like to have it a string. Is there an option to specify ...
0
votes
0answers
19 views
How to split text by symbol in Spark Streaming?
I'm trying to split text 2 times.
1st by | and then by ; in pyspark
lines = ssc.textFileStream("/mnt/hgfs/Shared/")
data = lines.map(lambda x: x.split("|"))
ndata = data.map(lambda ...
0
votes
0answers
54 views
pyspark foreach/foreachPartition send http request failed
I use urllib.request to send http request in foreach/foreachPartition. pyspark throw error as follow:
objc[74094]: +[__NSPlaceholderDate initialize] may have been in progress in another thread when ...