Hi Enrico,
Thanks for helping me to understand the mistakes.
My end goal is to achieve is something like the below:
1. Generate the data
2. Send the data in the batch of 4k records in one batch since the
API can accept the 4k records at once.
3. The record would be as the below:
4.
{
"Token": "", "CustomerName":"", "Object":"", "Data":
[{"A":"1"},{"A":"2"}]
}
5. Token will be generated first then it would be passed to the
'Token' key in the data.
For the above goal, I initially wrote something like the below which
gives a heap error because the data frame is getting created on the
driver side, and the size of the records is a minimum of 1M.
df = modifiedData
df = df.withColumn("uniqueID", lit("1"))
df = df.withColumn("row_num", row_number().over(
Window.partitionBy(col("uniqueID")).orderBy(col("uniqueID"))
))
tokenUrl = ""
# tokenUrl = ""
policyUrl = ""
tokenBody = {"Username": "", "Password": "",
"CustomerName": ""}
def get_token(url, payload):
try:_2022_06_11
print("Getting Token")
response = requests.request("POST", url, data=payload)
data = response.json()
if data['ErrorDescription'] == 'Success':
token = data['Token']
print(":::Token Generated::::")
else:
print('TokenNotGeneratedFrom: ')
# raise TokenNotGeneratedFrom(500, 'Token not
Generated from ')
return token
except Exception as e:
print('TokenNotGeneratedFrom: ' + str(e))
# raise TokenNotGeneratedFrom(500, str(e))
def call_to_cust_bulk_api(url, payload):
print("Calling Bulk API")
try:
# TODO: write code...
headers = {'content-type': 'application/json'}
print(":::::::jsn load::::")
# print(json.dumps(payload))
# print(payload)
response = requests.post
<http://requests.post>(url, data=json.dumps(payload), headers=headers)
# print(json.dumps(payload))
data = response.json()
return data
except Exception as e:
print('ExceptionInPushingDataTo: ' + str(e))
# raise ExceptionInPushingDataTo(500, str(e))
total_count = df.count()
i = 1
while i < total_count:
rangeNum = i + 3999
print("Range Num:::")
print(rangeNum)
df1 = df.filter((col("row_num") >= i) &
(col("row_num") <= rangeNum))
df1.cache()
maxValue =
df1.select(max(col("row_num")).alias("max_val")).toPandas()['max_val'].iloc[0]
finalDF = df1.drop("row_num", "edl_timestamp", "uniqueID")
print("finalDF count:::", finalDF.count())
token = get_token(tokenUrl, tokenBody)
result =
json.loads((finalDF.toPandas().to_json(orient="records")))
# token = get_token(tokenUrl, tokenBody)
custRequestBody = {
"Token": token,
"CustomerName": "",
"Object": "",
"Data": result
}
# print("::::Customer Request Body::::::")
# print(json.dumps(custRequestBody))
response = call_to_cust_bulk_api(policyUrl,
custRequestBody)
print(response)
finalDFStatus = finalDF.withColumn("edl_timestamp",
to_timestamp(lit(F.TimeNow()))).withColumn(
"status_for_each_batch",
lit(str(response)))
print("Max Value:::")
print(maxValue)
print("Next I:::")
i = rangeNum + 1
print(i)
This is my very first approach to hitting the APIs with Spark. So,
could you please help me to redesign the approach, or can share some
links or references using which I can go to the depth of this and
rectify myself. How can I scale this?
Any help is much appreciated. Thank you so much again for your time.
TIA,
Sid
On Fri, Jun 10, 2022 at 8:54 PM Enrico Minack <i...@enrico.minack.dev>
wrote:
Hi,
This adds a column with value "1" (string) *in all rows*:
|df = df.withColumn("uniqueID", lit("1")) |
||This counts the rows for all rows that have the same |uniqueID|,
*which are all rows*. The window does not make much sense.
And it orders all rows that have the same |uniqueID |by
|uniqueID|. Does not make much sense either.
|df = df.withColumn("row_num", row_number().over(
Window.partitionBy(col("uniqueID")).orderBy(col("uniqueID")) )) |
Then it looks like it takes the first 4000 rows (row_num from 1 to
4000) and tries to send them via HTTP POST. Then it moves the
range by one and sends row 2 to 4001 (mostly overlapped with the
first POST).
It is not clear if the "Data" field is meant to be all rows or
only a single row. Either way, this is not what happens. Please
consider the difference between a Column and a DataFrame in Spark.
This is very different from Pandas.
I think you have to rethink your approach. Using Spark means big
data. This approach is iterative and single-threaded.
Enrico
Am 10.06.22 um 16:01 schrieb Sid:
Hi Enrico,
Thanks for your time. Much appreciated.
I am expecting the payload to be as a JSON string to be a record
like below:
{"A":"some_value","B":"some_value"}
Where A and B are the columns in my dataset.
On Fri, Jun 10, 2022 at 6:09 PM Enrico Minack
<i...@enrico.minack.dev> wrote:
Sid,
just recognized you are using Python API here. Then
||struct(*colsListToBePassed))|| should be correct, given it
takes a list of strings.
Your method |call_to_cust_bulk_api| takes argument |payload|,
which is a ||Column||. This is then used in
|custRequestBody|. That is pretty strange use of a column
expression. What do you expect |print(payload)| to be?
I recommend to split that complex command into multiple
commands to find out what "an error of column not iterable"
refers to.
Enrico
Am 10.06.22 um 13:39 schrieb Enrico Minack:
Hi Sid,
||finalDF =
finalDF.repartition(finalDF.rdd.getNumPartitions())
.withColumn("status_for_batch",
call_to_cust_bulk_api(policyUrl,
to_json(struct(*colsListToBePassed)))) | |
You are calling ||withColumn|| with the result of
||call_to_cust_bulk_api|| as the second argument. That
result looks like it is of type string. But ||withColumn||
expects type ||Column||. You can turn that string into a
||Column|| using ||lit||:
||finalDF =
finalDF.repartition(finalDF.rdd.getNumPartitions())
.withColumn("status_for_batch",
lit(call_to_cust_bulk_api(policyUrl,
to_json(struct(*colsListToBePassed))))) ||
You are saying that gives you an error of column not
iterable. I reckon the ||struct(*colsListToBePassed))|| is
wrong.
Method ||struct|| requires a single string followed by a
list of strings. Given your ||colsListToBePassed|| is a list
of strings, this does not work. Try:
|| struct(||||||colsListToBePassed.head,
||colsListToBePassed.tail|||||: _*|))||
Alternatively, ||struct|| requires a list of ||Column||, so
try this:
|| struct(||||||colsListToBePassed.map(col)|||||||: _*|))||
The API is pretty clear about the types it expects.
If you are still having errors, you better please paste the
code and error.
Enrico
Am 09.06.22 um 21:31 schrieb Sid:
Hi Experts,
I am facing one problem while passing a column to the
method. The problem is described in detail here:
https://stackoverflow.com/questions/72565095/how-to-pass-columns-as-a-json-record-to-the-api-method-using-pyspark
TIA,
Sid
