小编Roe*_*ant的帖子

我在 AWS EMR 上使用 pyspark（4 个 r5.xlarge 作为 4 个工作线程，每个工作线程有 1 个执行程序和 4 个核心），并且我得到了AttributeError: Can't get attribute 'new_block' on <module 'pandas.core.internals.blocks'. 下面是引发此错误的代码片段：

search =  SearchEngine(db_file_dir = "/tmp/db")
conn = sqlite3.connect("/tmp/db/simple_db.sqlite")
pdf_ = pd.read_sql_query('''select  zipcode, lat, lng, 
                        bounds_west, bounds_east, bounds_north, bounds_south from 
                        simple_zipcode''',conn)
brd_pdf = spark.sparkContext.broadcast(pdf_) 
conn.close()


@udf('string')
def get_zip_b(lat, lng):
    pdf = brd_pdf.value 
    out = pdf[(np.array(pdf["bounds_north"]) >= lat) & 
              (np.array(pdf["bounds_south"]) <= lat) & 
              (np.array(pdf['bounds_west']) <= lng) & 
              (np.array(pdf['bounds_east']) >= lng) ]
    if len(out):
        min_index = np.argmin( (np.array(out["lat"]) - …

任何人都知道如何使用mysqldb将具有大量行的MySQL表转换为Python中的字典对象列表？

我的意思是将一组MySQL行(列'a','b'和'c')转换为如下所示的Python对象:

data = [ { 'a':'A', 'b':(2, 4), 'c':3.0 }, { 'a':'Q', 'b':(1, 4), 'c':5.0 }, { 'a':'T', 'b':(2, 8), 'c':6.1 } ]

谢谢 :)

在python 3中,我们可以使用以下脚本轻松地在同一行上打印.我用这个来理解我的循环的进度(剩下多少时间).但是,在jupyter中它不起作用(它在不同的行上打印)

import time
for f in range(10):
    print(f, end='\r', flush=True)
    time.sleep(10)

它不能打开%pprint的漂亮打印,我尝试使用sys.stdout.write(),但我也有这个问题.

测试以下代码时

@pytest.mark.asynico
async def test_handle_DATA(mocker):
    handle_mock = mocker.MagicMock()
    envelope_mock = mocker.MagicMock(mail_from="Test@From", rcpt_tos=["Test@To"], content=b"TestContent")

    result = SendToDictHandler.handle_DATA(handle_mock, "TestServer", "TestSession", envelope_mock)

    assert result == "250 Message accepted for delivery"
    assert email_core.testing_emails_dict == {
        "Test@To": {
            "from": "Test@From",
            "to": ["Test@To"],
            "msg": "TestContent",
        }
    }

pytest -vvv在项目环境中运行时收到的警告：

PytestWarning: Coroutine functions are not natively supported and have been skipped.
You need to install a suitable plugin for your async framework, for example:
 - pytest-asyncio
 - pytest-trio
 - pytest-tornasync
warnings.warn(PytestWarning(msg.format(pyfuncitem.nodeid)))

我确实pytest-asyncio安装了。我通过pytest --trace-config …

AWS Lambda 最近添加了从 docker 容器创建函数的选项（小警告：ECR 目前需要位于同一账户中）。我想知道：除了 docker 相对于随机 ZIP 的所有优点之外，这是否也可以更快地启动或更快地调用？

我们使用简单的日期时间版本对所有公司包进行版本控制。现在我们正在考虑转向pyproject.toml而不是setup.py. 是否也可以在那里进行灵活的版本控制？

version = datetime.datetime.now().strftime('%Y.%m.%d.%H%M')

# Actual setup
setup(
    name="some-package",
    version=version,
    description='Some description',
    packages=find_namespace_packages(where='src', include=['company.project.*']),
    package_dir={'': 'src'},
    python_requires='>=3.6',
    install_requires=[  
        'numpy',
        'numba'
    ],
)

我需要什么语法来调整 pyproject.toml 中的版本控制？这个正在使用poetry，但没有必要。

[tool.poetry]
name = "some-package"
version = "0.1.0"
description = ""
readme = "README.md"

我需要将基本功能的源代码存储在数据库中，并允许通过管理界面对其进行修改。这段代码将几个数字和字符串作为参数，并返回一个数字或None。我知道 eval 是邪恶的，所以我需要实现一种安全的方式来执行 Python 的一个非常基本的子集，或者至少在基于 Python 的 Web 应用程序中在语法上类似的东西。

显而易见的答案是实现 DSL（域特定语言），但是，我没有这方面的经验，也不知道从哪里开始，而且很多可用资源似乎有点超出我的想象。我希望也许已经有一些东西可以让我从数据库中的字符串基本上生成一个安全的 python 可调用函数。该语言实际上只需要支持赋值、基本数学、if/else 和不区分大小写的字符串比较。任何其他功能都是额外的，但我认为大多数事情都可以做到这一点，不需要复杂的数据结构、类、函数等。

如果目前不存在这样的东西，我愿意研究创建一个的可能性，但正如我所说，我不知道如何去做，在这方面的任何建议也将不胜感激。

我想从 AWS s3 存储桶流式传输我的数据文件。我遵循此处描述的设置，但使用tensorflow 2。

该设置指定您可以在 中使用 AWS 配置文件~/.aws/credentials，但我也尝试使用环境变量。然而下面的烟雾测试一直给出以下错误tensorflow.python.framework.errors_impl.UnimplementedError: File system scheme 's3' not implemented。

from tensorflow.python.lib.io import file_io
print(file_io.stat('s3://bucketname/key/'))

我已使用 手动添加了诗歌插件poetry self add xxx，但我没有看到 pyproject.toml 或诗歌.lock 中的任何行发生更改。我希望这个插件包含在开发环境中，因此需要在其他人设置环境时安装它。您可以将插件添加到--dev依赖项中吗？

我知道在python37中我们有一个新的api asyncio.get_running_loop()，它很容易使用，让我们在调用协程时不需要显式地传递eventloop。

我想知道我们是否可以使用任何方法在 python36 中获得相同的效果？

# which allows us coding conveniently with this api:
import asyncio

async def test():
    print("hello world !")

async def main():
    loop = asyncio.get_running_loop()
    loop.create_task(test())

asyncio.run(main())