pyspark dataframe joining of two dataframe












0















I have two dataframes say df1 and df2:
df1 has fields as CI_NAME,CLOSE_TIME,CH_ID
and df2 has fields as NAME,TIMESTAMP,MEM_CONSUMED.
Basically df1 has records of software updates done to the system and df2 has monitoring records of the system.



I need to add a field in df1 named cpu_util_avg_before_update by comparing CI_NAME equal to NAME field of df2 and CLOSE_TIME between TIMESTAMP - 7 days and TIMESTAMP and then take the average of MEM_CONSUMED.



How can I do that, any help would be appreciated as I have tried udf but that is not taking dataframe as input.
Thanks



here is the code that I tried:



from pyspark.sql.functions import col,udf,struct

from dateutil import parser

import datetime

@udf

def memavgbeforeupdate(structx,df2):
df=df2.where(col("name")==structx[1] & (col("timestamp")>parser.parse(structx[0])-datetime.timedelta(days=10) & col("timestamp")<parser.parse(structx[0])+datetime.timedelta(days=10)))

df=df.where(col("mem_consumed_average")!="NaN").where(col("mem_consumed_average").isNotNull())
if df.rdd.isEmpty():
return -1
else:
df1=df.select("mem_consumed_average")
return float(str(df1.select(mean(col("mem_consumed_average"))).collect()[0]).split("=")[1].split(")")[0])



df3=df1.withColumn("mem_avg_before_update",memavgbeforeupdate(struct(col("CLOSE_TIME"),col("CI_NAME")),df2))


But that's not working and throwing error as:




'DataFrame' object has no attribute '_get_object_id'











share|improve this question

























  • You cannot use DataFrame in udf. You'll have to rewrite this as a combination of joins and aggregations. Could you please provide edit your question and provide reproducible example?

    – user10465355
    Nov 22 '18 at 20:17
















0















I have two dataframes say df1 and df2:
df1 has fields as CI_NAME,CLOSE_TIME,CH_ID
and df2 has fields as NAME,TIMESTAMP,MEM_CONSUMED.
Basically df1 has records of software updates done to the system and df2 has monitoring records of the system.



I need to add a field in df1 named cpu_util_avg_before_update by comparing CI_NAME equal to NAME field of df2 and CLOSE_TIME between TIMESTAMP - 7 days and TIMESTAMP and then take the average of MEM_CONSUMED.



How can I do that, any help would be appreciated as I have tried udf but that is not taking dataframe as input.
Thanks



here is the code that I tried:



from pyspark.sql.functions import col,udf,struct

from dateutil import parser

import datetime

@udf

def memavgbeforeupdate(structx,df2):
df=df2.where(col("name")==structx[1] & (col("timestamp")>parser.parse(structx[0])-datetime.timedelta(days=10) & col("timestamp")<parser.parse(structx[0])+datetime.timedelta(days=10)))

df=df.where(col("mem_consumed_average")!="NaN").where(col("mem_consumed_average").isNotNull())
if df.rdd.isEmpty():
return -1
else:
df1=df.select("mem_consumed_average")
return float(str(df1.select(mean(col("mem_consumed_average"))).collect()[0]).split("=")[1].split(")")[0])



df3=df1.withColumn("mem_avg_before_update",memavgbeforeupdate(struct(col("CLOSE_TIME"),col("CI_NAME")),df2))


But that's not working and throwing error as:




'DataFrame' object has no attribute '_get_object_id'











share|improve this question

























  • You cannot use DataFrame in udf. You'll have to rewrite this as a combination of joins and aggregations. Could you please provide edit your question and provide reproducible example?

    – user10465355
    Nov 22 '18 at 20:17














0












0








0








I have two dataframes say df1 and df2:
df1 has fields as CI_NAME,CLOSE_TIME,CH_ID
and df2 has fields as NAME,TIMESTAMP,MEM_CONSUMED.
Basically df1 has records of software updates done to the system and df2 has monitoring records of the system.



I need to add a field in df1 named cpu_util_avg_before_update by comparing CI_NAME equal to NAME field of df2 and CLOSE_TIME between TIMESTAMP - 7 days and TIMESTAMP and then take the average of MEM_CONSUMED.



How can I do that, any help would be appreciated as I have tried udf but that is not taking dataframe as input.
Thanks



here is the code that I tried:



from pyspark.sql.functions import col,udf,struct

from dateutil import parser

import datetime

@udf

def memavgbeforeupdate(structx,df2):
df=df2.where(col("name")==structx[1] & (col("timestamp")>parser.parse(structx[0])-datetime.timedelta(days=10) & col("timestamp")<parser.parse(structx[0])+datetime.timedelta(days=10)))

df=df.where(col("mem_consumed_average")!="NaN").where(col("mem_consumed_average").isNotNull())
if df.rdd.isEmpty():
return -1
else:
df1=df.select("mem_consumed_average")
return float(str(df1.select(mean(col("mem_consumed_average"))).collect()[0]).split("=")[1].split(")")[0])



df3=df1.withColumn("mem_avg_before_update",memavgbeforeupdate(struct(col("CLOSE_TIME"),col("CI_NAME")),df2))


But that's not working and throwing error as:




'DataFrame' object has no attribute '_get_object_id'











share|improve this question
















I have two dataframes say df1 and df2:
df1 has fields as CI_NAME,CLOSE_TIME,CH_ID
and df2 has fields as NAME,TIMESTAMP,MEM_CONSUMED.
Basically df1 has records of software updates done to the system and df2 has monitoring records of the system.



I need to add a field in df1 named cpu_util_avg_before_update by comparing CI_NAME equal to NAME field of df2 and CLOSE_TIME between TIMESTAMP - 7 days and TIMESTAMP and then take the average of MEM_CONSUMED.



How can I do that, any help would be appreciated as I have tried udf but that is not taking dataframe as input.
Thanks



here is the code that I tried:



from pyspark.sql.functions import col,udf,struct

from dateutil import parser

import datetime

@udf

def memavgbeforeupdate(structx,df2):
df=df2.where(col("name")==structx[1] & (col("timestamp")>parser.parse(structx[0])-datetime.timedelta(days=10) & col("timestamp")<parser.parse(structx[0])+datetime.timedelta(days=10)))

df=df.where(col("mem_consumed_average")!="NaN").where(col("mem_consumed_average").isNotNull())
if df.rdd.isEmpty():
return -1
else:
df1=df.select("mem_consumed_average")
return float(str(df1.select(mean(col("mem_consumed_average"))).collect()[0]).split("=")[1].split(")")[0])



df3=df1.withColumn("mem_avg_before_update",memavgbeforeupdate(struct(col("CLOSE_TIME"),col("CI_NAME")),df2))


But that's not working and throwing error as:




'DataFrame' object has no attribute '_get_object_id'








apache-spark dataframe pyspark data-science






share|improve this question















share|improve this question













share|improve this question




share|improve this question








edited Nov 25 '18 at 18:33







Neeraj Kumar

















asked Nov 22 '18 at 12:52









Neeraj KumarNeeraj Kumar

62




62













  • You cannot use DataFrame in udf. You'll have to rewrite this as a combination of joins and aggregations. Could you please provide edit your question and provide reproducible example?

    – user10465355
    Nov 22 '18 at 20:17



















  • You cannot use DataFrame in udf. You'll have to rewrite this as a combination of joins and aggregations. Could you please provide edit your question and provide reproducible example?

    – user10465355
    Nov 22 '18 at 20:17

















You cannot use DataFrame in udf. You'll have to rewrite this as a combination of joins and aggregations. Could you please provide edit your question and provide reproducible example?

– user10465355
Nov 22 '18 at 20:17





You cannot use DataFrame in udf. You'll have to rewrite this as a combination of joins and aggregations. Could you please provide edit your question and provide reproducible example?

– user10465355
Nov 22 '18 at 20:17












0






active

oldest

votes











Your Answer






StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");

StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);

StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});

function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});


}
});














draft saved

draft discarded


















StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53431481%2fpyspark-dataframe-joining-of-two-dataframe%23new-answer', 'question_page');
}
);

Post as a guest















Required, but never shown

























0






active

oldest

votes








0






active

oldest

votes









active

oldest

votes






active

oldest

votes
















draft saved

draft discarded




















































Thanks for contributing an answer to Stack Overflow!


  • Please be sure to answer the question. Provide details and share your research!

But avoid



  • Asking for help, clarification, or responding to other answers.

  • Making statements based on opinion; back them up with references or personal experience.


To learn more, see our tips on writing great answers.




draft saved


draft discarded














StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53431481%2fpyspark-dataframe-joining-of-two-dataframe%23new-answer', 'question_page');
}
);

Post as a guest















Required, but never shown





















































Required, but never shown














Required, but never shown












Required, but never shown







Required, but never shown

































Required, but never shown














Required, but never shown












Required, but never shown







Required, but never shown







Popular posts from this blog

404 Error Contact Form 7 ajax form submitting

How to know if a Active Directory user can login interactively

TypeError: fit_transform() missing 1 required positional argument: 'X'