Google Cloud Composer using google-cloud-bigquery python client library

The name of the pictureThe name of the pictureThe name of the pictureClash Royale CLAN TAG#URR8PPP

Google Cloud Composer using google-cloud-bigquery python client library



I'm trying to run a DAG in Google Cloud Composer in which the first component is to use a http GET request to call an API and then use the python-client library to insert the json into a BigQuery table. I am trying to run this function: https://googlecloudplatform.github.io/google-cloud-python/latest/bigquery/generated/google.cloud.bigquery.client.Client.insert_rows_json.html


import requests
import datetime
import ast
import numpy as np
from airflow import models
from airflow.contrib.operators import bigquery_operator
from airflow.operators import python_operator
import google.cloud.bigquery as bigquery

client = bigquery.Client(project = 'is-flagship-data-api-sand')
dataset_id = 'Mobile_Data_Test'
dataset_ref = client.dataset(dataset_id)
table_ref = dataset_ref.table('sample_wed')
table = client.get_table(table_ref)

def get_localytics_data():
profiles_requests_command = "https://%s:%s@api.localytics.com/v1/exports/profiles/%d/profile"%(api_key,api_secret,28761)
res_profiles = requests.get(profiles_requests_command)
if res_profiles.status_code == 200:
data = res_profiles.content
data_split = data.split('n')[:-1]
data_split_ast = [ast.literal_eval(x) for x in data_split]

#take out characters from the beginning to have neat columns
data_split_ast_pretty = [dict(zip(map(lambda x: x[4:], item.keys()), item.values())) for item in data_split_ast]


#add current date
current_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
for item in data_split_ast_pretty:
item['DateCreated'] = current_time


random_sample = list(np.random.choice(data_split_ast_pretty,5))
print random_sample
client.insert_rows_json(table = table, json_rows = random_sample)
else:
pass




run_api = python_operator.PythonOperator(task_id='call_api',
python_callable=get_localytics_data)



I added the PYPI Packages of :



requests ===2.19.1



numpy ===1.12.0



google-cloud-bigquery ===1.4.0



I get the error of : Broken DAG: [/home/airflow/gcs/dags/composer_test_july30_v2.py] 'Client' object has no attribute 'get_table'
in the Airflow UI Console.



All the code shown works locally but will not work using Cloud Composer.





What version of Cloud Composer are you using? You can get this from describing the environment.
– Trevor Edwards
17 mins ago









By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.

Comments

Post a Comment

Popular posts from this blog

Executable numpy error

Image
Clash Royale CLAN TAG #URR8PPP Executable numpy error I'm trying to create an executable file for my code. I've already tried with cx_Freeze and with pyinstaller. Both gives me the same error, which is: "Missing required dependencies 0.format(missing_dependencies))" PS C:UsersGustavoDesktopbuildexe.win32-3.6> python AgendaOficial.py C:UsersGustavoAppDataLocalProgramsPythonPython36-32python.exe: can't open file 'AgendaOficial.py': [Errno 2] No such file or directory PS C:UsersGustavoDesktopbuildexe.win32-3.6> .AgendaOficial.exe Traceback (most recent call last): File "C:UsersGustavoAppDataLocalProgramsPythonPython36-32libsite-packagescx_Freezeinitscripts__startup__.py", line 14, in run module.run() File "C:UsersGustavoAppDataLocalProgramsPythonPython36-32libsite-packagescx_FreezeinitscriptsConsole.py", line 26, in run exec(code, m. dict ) File "AgendaOficial.py", line 8, in File "C:UsersGustavoAppDataLocal...
Read more

PySpark count values by condition

Image
Clash Royale CLAN TAG #URR8PPP PySpark count values by condition I have a DataFrame, a snippet here: [['u1', 1], ['u2', 0]] basically one string ('f') and either a 1 or a 0 for second element ('is_fav'). What I need to do is grouping on the first field and counting the occurrences of 1s and 0s. I was hoping to do something like num_fav = count((col("is_fav") == 1)).alias("num_fav") num_nonfav = count((col("is_fav") == 0)).alias("num_nonfav") df.groupBy("f").agg(num_fav, num_nonfav) It does not work properly, I get in both cases the same result which amounts to the count for the items in the group, so the filter (whether it is a 1 or a 0) seems to be ignored. Does this depend on how count works? count 1 Answer 1 There is no filter here. Both col("is_fav") == 1 and col("is_fav") == 0) are just boolean expressions and count doesn't really care about their value as long a...
Read more

Mass disable jenkins jobs

Image
Clash Royale CLAN TAG #URR8PPP Mass disable jenkins jobs Is it possible to disable a large number of Jenkins jobs in one go? I have a large number of jenkins job I need to disable. Visiting each job in turn and disbling is tedious. 2 Answers 2 https://wiki.jenkins-ci.org/display/JENKINS/Configuration+Slicing+Plugin That works beautifully install it go to http://your_jenkins.com:8080/slicing/jobdisabledbool/? By using configuration slicing plugin: By using this plugin easy to find out anything has enabled or disable jobs and timer separated jobs Manage Jenkins >Config slicing > select options By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.
Read more