我试图弄清楚你需要在哪里使用一个lit值,这个值literal column在文档中被定义为.
以此为例udf,它返回SQL列数组的索引:
def find_index(column, index):
return column[index]
如果我要将一个整数传入此中,我会收到错误.我需要将lit(n)值传递给udf以获取数组的正确索引.
有没有我可以更好地学习的时候用硬性规定的地方lit和可能col呢？
我正在创建一个解析大量服务器数据的工作,然后将其上传到Redshift数据库中.
我的工作流程如下:
dataframes或spark sql来解析数据并写回S3我已经开始讨论如何自动执行此操作,以便我的进程旋转EMR集群,引导正确的安装程序,并运行我的python脚本,其中包含用于解析和编写的代码.
有没有人有任何可以与我分享的示例,教程或经验,以帮助我学习如何做到这一点？
根据PEP 8(最大行长度),行不应超过79个字符.
但是,当我尝试拆分查询时,会遇到连续字符和无效标记等问题.
例如,根据PEP8,格式化此查询的最佳方法是什么？
cursor.execute("SELECT pivot_id FROM aud_qty WHERE hshake1 is NULL AND ((strftime('%s', DATETIME('now')) - strftime('%s', sent_to_pivot)) / (60)) > 30;")
我正在拆分HTTP请求以查看元素,我想知道是否有一种方法可以指定我想在同一个调用中查看的元素而无需执行其他操作.
例如:
from pyspark.sql import functions as fn
df.select(fn.split(df.http_request, '/').alias('http'))
给我一个新Dataframe的数组行如下:
+--------------------+
| http|
+--------------------+
|[, courses, 26420...|
我想要索引1(课程)中的项目,而不必再做另一个select语句指定df.select(df.http[1])或其他.这可能吗？
我正在编写我的第一个Django应用程序,跟着这本书:
在书中有一个测试,验证html是否按预期返回.这是测试:
def test_home_page_returns_correct_html(self):
request = HttpRequest()
response = home_page(request)
expected_html = render_to_string('home.html')
print(expected_html)
print(response.content.decode())
self.assertEqual(response.content.decode(), expected_html)
我的测试在测试中失败,assertEqual因为我csrf token在我的HTML中添加了一个Django Template Language.这是我的HTML页面的样子:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>To-Do lists</title>
</head>
<body>
<h1>Your To-Do list</h1>
<form method="POST">
<input name="item_text" id="id_new_item" placeholder="Enter a to-do item"/>
{% csrf_token %}
</form>
<table id="id_list_table">
<tr><td>{{ new_item_list }}</td></tr>
</table>
</body>
</html>
由于render_to_string方法不包括令牌,我的断言失败.以下是print我的测试中包含的两个陈述:
F<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>To-Do lists</title>
</head> …我正在研究从Redshift数据库转储文件的过程,并且不希望本地下载文件来处理数据.我看到Java有一个StreamingObject类可以做我想要的,但我没有看到类似的东西boto3.
我正在创建一个程序,它将提取一个帐号列表,然后运行一个ls -lh命令来查找每个帐号.当我在没有Python的Linux服务器上运行我的命令时,它会提取文件没问题,但是当我通过Python执行它时它说它无法找到它们.
import subprocess as sp
sp.call(['cd', input_dir])
for i, e in enumerate(piv_id_list):
proc_out = sp.Popen(['ls', '-lh', '*CSV*APP*{0}.zip'.format(e)])
proc_out_list.append(proc_out)
print(proc_out)
这是通过Python解释器运行命令时的一些示例输出:
>>> ls:无法访问*CSV1000*APP*:没有这样的文件或目录
但是通过Linux同样的命令:
ls -lh *CSV*APP*
它会像它应该的那样返回输出.
我正在使用logging模块，并且传递了与当前正在执行的其他作业相同的参数：
import logging
from inst_config import config3
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] - %(message)s',
filename=config3.GET_LOGFILE(config3.REQUESTS_USAGE_LOGFILE))
logging.warning('This should go in the file.')
if __name__ == '__main__':
logging.info('Starting unload.')
使用此方法创建文件名：
REQUESTS_USAGE_LOGFILE = r'C:\RunLogs\Requests_Usage\requests_usage_runlog_{}.txt'.format(
CUR_MONTH)
def GET_LOGFILE(logfile):
"""Truncates file and returns."""
with open(logfile, 'w'):
pass
return logfile
但是，当我运行它时，它正在创建文件，然后仍将日志记录信息输出到控制台。我正在跑步Powershell。
只是尝试将其放在主语句中，如下所示：
if __name__ == '__main__':
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] - %(message)s',
filename=config3.GET_LOGFILE(config3.REQUESTS_USAGE_LOGFILE))
logging.warning('This should go in the file.')
仍然没有运气。
我们存储了大量的服务器数据S3(很快就会采用Parquet格式化).数据需要一些转换,因此它不能是S3的直接副本.我将Spark用来访问数据,但我想知道是不是用Spark操纵它,写回S3,然后复制到Redshift如果我可以跳过一步并运行查询来拉/变换数据,然后直接复制到Redshift？
我有一个字典，其中每个键都是值的元组，我想使用该sorted()方法在元组的第一个元素上对字典进行排序。我的代码如下所示：
def mapData(header_list, dict_obj):
master_dict = {}
client_section_list = []
for element in header_list:
for row in dict_obj:
if (row['PEOPLE_ID'], row['DON_DATE']) == element:
client_section_list.append(row)
element = list(element)
element_list = [client_section_list[0]['DEDUCT_AMT'],
client_section_list[0]['ND_AMT'],
client_section_list[0]['DEDUCT_YTD'],
client_section_list[0]['NONDEDUCT_YTD']
]
try:
element_list.append((float(client_section_list[0]['DEDUCT_YTD']) +
float(client_section_list[0]['NONDEDUCT_YTD'])
))
except ValueError:
pass
element.extend(element_list)
element = tuple(element)
master_dict[element] = client_section_list
client_section_list = []
return sorted(master_dict, key=lambda key: key[master_dict[(1)]]
最后一行是我试图找到一种对其进行排序的方法。我的元组看起来像这样：
(312178078,6/22/15,25,0,25,0,25.0)
python ×9
apache-spark ×4
amazon-s3 ×3
pyspark ×3
python-2.7 ×2
amazon-emr ×1
boto3 ×1
dictionary ×1
django ×1
hadoop ×1
html5 ×1
logging ×1
pep8 ×1
powershell ×1
python-2.6 ×1
sorting ×1
subprocess ×1
windows ×1