Questions tagged [pyspark]
The Spark Python API (PySpark) exposes the apache-spark programming model to Python.
33,174
questions
0
votes
0
answers
6
views
Looking to get counts of items within ArrayType column without using Explode
NOTE: I'm working with Spark 2.4
Here is my dataset:
df
col
[1,3,1,4]
[1,1,1,2]
I'd like to essentially get a value_counts of the values in the array. The results df wou
df_upd
col
[{1:2},{3:1},{4:1}]...
0
votes
0
answers
9
views
Spark dataframe column resists renaming
I have a Spark dataframe df with exactly one column named "My Column Name". It's created by reading in a parquet file. The printSchema function returns this:
>>> df.printSchema()
...
1
vote
0
answers
15
views
Pysark out of memory issues with large dataset
i got this error:
An error occurred while calling o154.save. Job aborted due to stage failure: Total size of serialized results of 781 tasks (2.0 GiB) is bigger than spark.driver.maxResultSize (2.0 ...
-1
votes
0
answers
9
views
Infer JSON Schema from a Spark Dataframe String Column
I have a spark dataframe with a single string column storing json values.
I need to infer the json schema from that string column and create another dataframe based on that json schema with multiple ...
0
votes
0
answers
23
views
Cannot resolve stack() due to type mismatch
I have a pyspark code that looks like this:
from pyspark.sql.functions import expr
unpivotExpr = """stack(14, 'UeEnd', UeEnd,
'Encedreco', Endereco,
'UeSitFun', UeSitFun,
'...
0
votes
0
answers
6
views
org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException pyspark
I am running pyspark locally on my mac 16GB with 6 cores and I am reading a json file and converting it to parquet.I am receiving this error only for one file, rest of files pyspark is able to read ...
0
votes
0
answers
8
views
How to automate casting of empty array<string> elements to array<struct<>> elements in pyspark
We have a LogicalDate as column in a pyspark dataframe, having array of struct type schema
Data:
{
"LogicalLinks": [
{
"ClassId": "myclassId",
...
2
votes
0
answers
13
views
Access Denied when performing S3 Multipart Upload operation into SSE-KMS encrypted bucket
I'm facing Access Denied for Multipart Upload into SSE-KMS encrypted bucket.
Code is running Glue (might be from other services, have no ability to verify). I’ve tried set of different permissions, ...
1
vote
1
answer
17
views
Assign list of values to columns with binary value in PySpark
I have a pyspark dataframe with a column Categories that has a list of values, for a given ID. Looks like this, let's call this df1:
ID | Categories
122 [16777277, 16777350, 16777483, 16777425, ...
0
votes
1
answer
10
views
Overlapping polygons with hexagons
I have been struggling with polygons in python 3 for a long time. There is a geojson with a list of polygons and multipolygons. It is necessary to cover these polygons with hexes of a CERTAIN level. ...
0
votes
0
answers
5
views
Great expectations v3 API in aws glue 3.0
I'm trying to a validation in the pipeline using Great expectations on AWS glue 3.0.
Here's my initial attempt to create the data context at runtime based on their docs
def create_context():
...
0
votes
1
answer
14
views
Calculate the number of rows for a list of hive tables with pyspark
I have created a spark program that retrieves the table names from a Hive database and then calculates the row count for a single table. However, I am trying to level this up by getting the row count ...
0
votes
1
answer
13
views
Reading an Elasticsearch Index from PySpark
Could anyone tell me why this test script for PySpark errors out?
(python 3.6.8, hadoop 3.3.1, spark 3.2.1, elasticsearch-hadoop 7.14)
from pyspark.sql import SparkSession, SQLContext
myspark = ...
0
votes
0
answers
16
views
How to look if a value is in a group of columns in Pyspark
I've a big dataframe with multiple columns, some of those columns are called col1:col35.
I'd like to know if a specif value is contained in ANY of those columns.
If 'RP' is in (col1:col35) then val= 1;...
3
votes
1
answer
34
views
How to create new column dynamically in pandas like we do in pyspark withColumn
from statistics import mean
import pandas as pd
df = pd.DataFrame(columns=['A', 'B', 'C'])
df["A"] = [1, 2, 3, 4, 4, 5, 6]
df["B"] = ["Feb", "Feb", "Feb&...
0
votes
0
answers
16
views
Pandas on spark dataframe assign function ValueError
I want to create a new column in a pandas-on-spark dataframe. In previous attempts, I tried pure pandas syntax like `df['column']=...' but I struggle against pickling error which I could not ...
0
votes
0
answers
15
views
Differentiate sql null and json null pyspark dataframe
I have a json file with inconsistent schema in which some fields may or may not present in the successive rows
Sample JSON file
{"table":"TABLEA","ID":1,"COLUMN1&...
1
vote
0
answers
10
views
How to run Mosaic locally (outside Databricks)
I am using Mosaic on Databricks to do some geospatial transformations, and so far it works well.
As my code base starts growing, I am looking for a way to run unit tests on the geospatial ...
-4
votes
0
answers
26
views
remove square bracket using regex on spark [duplicate]
I have a regex pattern and to that I want to make changes such that the code ignores square brackets"[]"
This is my regex pattern "\,|\\s+|\W:|="
What changes do I need to make so ...
0
votes
0
answers
10
views
Status of structured streaming query in PySpark
I am following the book Spark - Definitive Guide and I was writing basic program that streams the data . The books says that I should use awaitTermination() method to process the query correctly. When ...
0
votes
0
answers
15
views
Making all String Dates Uniform to YYYYMMDD in PySpark
i've got a pyspark dataframe with about 60 columns
|name|address|date|
| han|123. |19970401|
|dan |boul. |20100509|
|lan |street1|2010-04-02|
|kim |street2|02112020|
|red |street3||
|derek|street4|...
0
votes
1
answer
18
views
JSON to DataFrame in Spark, return value is null
I have a key value pair where value is JSON.
xyz= (null, "{"ID": "1", "Invoice No": "XYZ-123", "Invoice Date": "22/8/2022", "...
0
votes
0
answers
8
views
How to create a managed Hive Table using pyspark
I am facing the problem that every table I create using pyspark has type EXTERNAL_TABLE in hive, but I want to create managed tables and don't know what I am doing wrong. I tried different ...
1
vote
1
answer
31
views
PySpark: how to get all combinations of columns
I have a DF with combinations of batches, inputs and outputs and I would like to be able to add their "unique combinations" back to the DataFrame. A simple representation of the data looks ...
0
votes
0
answers
11
views
PySpark partition by to Pandas
I have a labled_features dataframe which looks like this (sample)-
datetime machineID error1_RM24 failure
0 2021-01-01 00:00:00 0.000317 1 0.000367 0....
0
votes
1
answer
15
views
Execute spark sql query on dataframe column values
I'm trying to get the size of each table in my database.
I listed first all my tables in a dataframe using this command :
df = spark.sql("show tables in db")
And this is my current ...
-1
votes
2
answers
28
views
Why does a join add duplicate column names
I have two tables and when I do a join it joins the tables together side by side not doing row based if same columns and not sure why this could be.
+--------+--------+--------+
| VALUE| DAY | ...
-1
votes
0
answers
20
views
Hello, in the code I gave below, TypeError: Method __init__ forces keyword arguments. I'm getting the error, what do you think is the reason?
Hello, in the code I gave below, TypeError: Method __ init __ forces keyword arguments. I'm getting the error, what do you think is the reason?
The purpose of the code is to predict whether the ...
0
votes
1
answer
26
views
How do I receive a subset of data from Kafka?
I know we can filter the data in our consumer/streaming programs. But what I am looking for here is a solution to filter the data from Kafka broker itself.
Problem Statement: I have a use case to ...
0
votes
0
answers
11
views
Trigerring spark-submit in local mode from a remote server
I have a pyspark application that i am currently running in local mode. Going forward this needs to be deployed in production and the pyspark job needs to be trigerred from clients end.
What changes ...
0
votes
0
answers
17
views
What is the best way for data movement between seperate dedicated sql pools in synapse?
Can anyone tell me the best way to read data from synapse dedicated sql pool tables in synapse notebooks? My requirement is to read selected columns from tables present in two separate dedicated pools,...
0
votes
0
answers
14
views
Perform preprocessing operations from pandas on spark dataframe
I have a rather large csv so I am using AWS EMR to read the data into a spark dataframe to perform some operations. I have a pandas function that does some simple preprocessing
def clean_census_data(...
0
votes
1
answer
26
views
Pyspark udf to detect "Actors"
I have a matrix(dataframe) I want to find all the rows there the row and columns intersect with a '1'. (The 'Character' row value matches the column name)
Example. Sam is an actor. (He has a '1' in ...
0
votes
1
answer
27
views
Remove everything after a space - Pyspark
I have a dataframe df as follows:
A B
21k2 b 1
2412 9 p
Both A and B are strings.
I would like for the A column elements to be trimmed as follows:
A ...
-1
votes
2
answers
20
views
Spark sql function to get time stamp from date time string
I am converting sql version of a code to spark-sql (pyspark) one. I need to get time component from a datetime string. In sql, the code used is
to_char(a.timeclose,''HH24:MI:SS'') as timeclose
Which ...
0
votes
0
answers
20
views
Using Spark in Kubernetes from app/driver outside the cluster
Context
I'm trying to configure Spark in Kubernetes to run dynamic executors in the cluster, with the app/driver being outside the cluster. My knowledge level is pretty high on Kubernetes, but very ...
0
votes
0
answers
15
views
How to convert PySpark df from nested boolean columns to filtered array of struct?
I have a df_have with this schema:
root
|-- events: struct (nullable = true)
| |-- eventA: boolean (nullable = true)
| |-- eventB: boolean (nullable = true)
| |-- eventC: boolean (...
0
votes
0
answers
9
views
Class org.apache.hadoop.fs.s3a.S3AFileSystem not found ( Local pyspark submit )
I am using spark 3.2.1
Hadoop version =3.3.1
(sc._gateway.jvm.org.apache.hadoop.util.VersionInfo.getVersion())
I saw in other forums that this error is received because of incorrect pyspark packages.
...
0
votes
0
answers
7
views
Why is applying a model to a batch of data in Spark via mapPartitions() much faster than applying to individual rows via map()?
I'm trying to learn Spark and especially it's functionality mapPartitions().
Let's say we have a dataset of 1000 datapoints.
If you apply batch processing by chunking the data into 10 partitions then ...
2
votes
1
answer
31
views
In Palantir Foundry how do I parse a very large csv file with OOMing the driver or executor?
Similar to How do I parse large compressed csv files in Foundry? but without the file being compressed, a system generated (>10GB) csv file which needs to be parsed as a Foundry Dataset.
A dataset ...
0
votes
1
answer
13
views
Pyspark : Convert Julian Date to Calendar date
I have a pySpark DataFrame Column with Julian Dates. I tried to convert the date to Calender Date.
number
julian_date
1
17196
2
17199
3
17281
I tried with the below code:
spdf = spdf.withColumn('...
0
votes
0
answers
14
views
PySpark DataFrame not showing in spark environment
So I was able to initialize a spark cluster and pull in data from a database using PySpark, but I was running into the following issue. When performing df.show(n=10) it was taking a super long time to ...
1
vote
0
answers
9
views
How to upload embeddings into Milvus?
Imagine I have a Spark DataFrame data with more than 10000 rows:
id
text
1
Mercury is the closest planet to the Sun. The smallest planet in the Solar System, Mercury has no natural satellites. The ...
1
vote
1
answer
15
views
SQL/Pyspark query to compre each value in a group with an aggregation of the same group
Suppose I have a table like this
id
date
value
1
01/04
7
1
01/03
2
1
01/02
10
1
01/01
4
2
01/04
7
2
01/03
7
2
01/02
8
2
01/01
5
What I want to get is another table which the same number of ...
0
votes
0
answers
21
views
ValueError: Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame
I am trying to check either a element is existing in the list or not. And I am getting above mentioned error. Last line of the given code where 'if' condition pointed is throwing this error.
...
1
vote
1
answer
16
views
How does spark calculates number of records in a dataframe?
I know that df.count() will trigger a spark action and return number of records present in a dataframe, but I wanted to know how this process work internally does spark goes through the whole ...
0
votes
1
answer
17
views
Installing Apache Spark Packages to run Locally
I am looking for a clear guide or steps to installing Spark packages (specifically spark-avro) to run locally and correctly using them with spark-submit command.
I've spent a lot of time reading many ...
0
votes
1
answer
29
views
How to change the data type of the columns of a PySpark dataframe?
I have a dataframe where all the columns are of type string and I need them to be of type double. I have code that does it:
df_Double = df.select([df[c].cast(DoubleType()).alias(c) for c in df.columns]...
0
votes
0
answers
21
views
Spark Dataframe containing single column, convert all rows into a single row and then create a single json out of that row
----------------------------------------------------------------------------------------------+
|value |
+-------...
2
votes
1
answer
29
views
How to use Apache Sedona on Databricks Delta Live tables?
I am trying to run some geospatial transformations in Delta Live Table, using Apache Sedona. I tried defining a minimal example pipeline demonstrating the problem I encounter.
First cell of my ...