diff --git a/docs/snippets.py b/docs/snippets.py
index 8171d5cf8..16b324023 100644
--- a/docs/snippets.py
+++ b/docs/snippets.py
@@ -51,8 +51,7 @@ def storage_get_started(client, to_delete):
 @snippet
 def client_bucket_acl(client, to_delete):
     bucket_name = "system-test-bucket"
-    bucket = client.bucket(bucket_name)
-    bucket.create()
+    client.create_bucket(bucket_name)
 
     # [START client_bucket_acl]
     client = storage.Client()
diff --git a/google/cloud/storage/bucket.py b/google/cloud/storage/bucket.py
index 4914844dd..fae43104f 100644
--- a/google/cloud/storage/bucket.py
+++ b/google/cloud/storage/bucket.py
@@ -693,7 +693,7 @@ def create(
         predefined_default_object_acl=None,
         timeout=_DEFAULT_TIMEOUT,
     ):
-        """Creates current bucket.
+        """DEPRECATED. Creates current bucket.
 
         If the bucket already exists, will raise
         :class:`google.cloud.exceptions.Conflict`.
@@ -737,44 +737,24 @@ def create(
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
         """
+        warnings.warn(
+            "Bucket.create() is deprecated and will be removed in future."
+            "Use Client.create_bucket() instead.",
+            PendingDeprecationWarning,
+            stacklevel=1,
+        )
         if self.user_project is not None:
             raise ValueError("Cannot create bucket with 'user_project' set.")
 
         client = self._require_client(client)
-
-        if project is None:
-            project = client.project
-
-        if project is None:
-            raise ValueError("Client project not set:  pass an explicit project.")
-
-        query_params = {"project": project}
-
-        if predefined_acl is not None:
-            predefined_acl = BucketACL.validate_predefined(predefined_acl)
-            query_params["predefinedAcl"] = predefined_acl
-
-        if predefined_default_object_acl is not None:
-            predefined_default_object_acl = DefaultObjectACL.validate_predefined(
-                predefined_default_object_acl
-            )
-            query_params["predefinedDefaultObjectAcl"] = predefined_default_object_acl
-
-        properties = {key: self._properties[key] for key in self._changes}
-        properties["name"] = self.name
-
-        if location is not None:
-            properties["location"] = location
-
-        api_response = client._connection.api_request(
-            method="POST",
-            path="/b",
-            query_params=query_params,
-            data=properties,
-            _target_object=self,
+        client.create_bucket(
+            bucket_or_name=self,
+            project=project,
+            location=location,
+            predefined_acl=predefined_acl,
+            predefined_default_object_acl=predefined_default_object_acl,
             timeout=timeout,
         )
-        self._set_properties(api_response)
 
     def patch(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Sends all changed properties in a PATCH request.
@@ -1908,7 +1888,7 @@ def requester_pays(self):
     def requester_pays(self, value):
         """Update whether requester pays for API requests for this bucket.
 
-        See  https://cloud.google.com/storage/docs/<DOCS-MISSING> for
+        See https://cloud.google.com/storage/docs/using-requester-pays for
         details.
 
         :type value: convertible to boolean
diff --git a/google/cloud/storage/client.py b/google/cloud/storage/client.py
index 41c123880..ed34ec16f 100644
--- a/google/cloud/storage/client.py
+++ b/google/cloud/storage/client.py
@@ -14,8 +14,8 @@
 
 """Client for interacting with the Google Cloud Storage API."""
 
+import warnings
 import functools
-
 import google.api_core.client_options
 
 from google.auth.credentials import AnonymousCredentials
@@ -373,8 +373,9 @@ def create_bucket(
             ]):
                 The bucket resource to pass or name to create.
             requester_pays (bool):
-                Optional. Whether requester pays for API requests for this
-                bucket and its blobs.
+                DEPRECATED. Use Bucket().requester_pays instead.
+                Optional. Whether requester pays for API requests for
+                this bucket and its blobs.
             project (str):
                 Optional. The project under which the bucket is to be created.
                 If not passed, uses the project set on the client.
@@ -435,6 +436,11 @@ def create_bucket(
             raise ValueError("Client project not set:  pass an explicit project.")
 
         if requester_pays is not None:
+            warnings.warn(
+                "requester_pays arg is deprecated. Use Bucket().requester_pays instead.",
+                PendingDeprecationWarning,
+                stacklevel=1,
+            )
             bucket.requester_pays = requester_pays
 
         query_params = {"project": project}
diff --git a/tests/unit/test_bucket.py b/tests/unit/test_bucket.py
index 312fc0f65..301dc61a1 100644
--- a/tests/unit/test_bucket.py
+++ b/tests/unit/test_bucket.py
@@ -615,223 +615,6 @@ def api_request(cls, *args, **kwargs):
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
 
-    def test_create_w_user_project(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        USER_PROJECT = "user-project-123"
-
-        client = Client(project=PROJECT)
-        client._base_connection = _Connection()
-
-        bucket = self._make_one(client, BUCKET_NAME, user_project=USER_PROJECT)
-
-        with self.assertRaises(ValueError):
-            bucket.create()
-
-    def test_create_w_missing_client_project(self):
-        from google.cloud.storage.client import Client
-
-        BUCKET_NAME = "bucket-name"
-
-        client = Client(project=None)
-        bucket = self._make_one(client, BUCKET_NAME)
-
-        with self.assertRaises(ValueError):
-            bucket.create()
-
-    def test_create_w_explicit_project(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        OTHER_PROJECT = "other-project-123"
-        DATA = {"name": BUCKET_NAME}
-        connection = _make_connection(DATA)
-
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-
-        bucket = self._make_one(client, BUCKET_NAME)
-        bucket.create(project=OTHER_PROJECT)
-        connection.api_request.assert_called_once_with(
-            method="POST",
-            path="/b",
-            query_params={"project": OTHER_PROJECT},
-            data=DATA,
-            _target_object=bucket,
-            timeout=self._get_default_timeout(),
-        )
-
-    def test_create_w_explicit_location(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        LOCATION = "us-central1"
-        DATA = {"location": LOCATION, "name": BUCKET_NAME}
-
-        connection = _make_connection(
-            DATA, "{'location': 'us-central1', 'name': 'bucket-name'}"
-        )
-
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-
-        bucket = self._make_one(client, BUCKET_NAME)
-        bucket.create(location=LOCATION)
-
-        connection.api_request.assert_called_once_with(
-            method="POST",
-            path="/b",
-            data=DATA,
-            _target_object=bucket,
-            query_params={"project": "PROJECT"},
-            timeout=self._get_default_timeout(),
-        )
-        self.assertEqual(bucket.location, LOCATION)
-
-    def test_create_hit(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        DATA = {"name": BUCKET_NAME}
-        connection = _make_connection(DATA)
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-
-        bucket = self._make_one(client=client, name=BUCKET_NAME)
-        bucket.create(timeout=42)
-
-        connection.api_request.assert_called_once_with(
-            method="POST",
-            path="/b",
-            query_params={"project": PROJECT},
-            data=DATA,
-            _target_object=bucket,
-            timeout=42,
-        )
-
-    def test_create_w_extra_properties(self):
-        from google.cloud.storage.client import Client
-
-        BUCKET_NAME = "bucket-name"
-        PROJECT = "PROJECT"
-        CORS = [
-            {
-                "maxAgeSeconds": 60,
-                "methods": ["*"],
-                "origin": ["https://example.com/frontend"],
-                "responseHeader": ["X-Custom-Header"],
-            }
-        ]
-        LIFECYCLE_RULES = [{"action": {"type": "Delete"}, "condition": {"age": 365}}]
-        LOCATION = "eu"
-        LABELS = {"color": "red", "flavor": "cherry"}
-        STORAGE_CLASS = "NEARLINE"
-        DATA = {
-            "name": BUCKET_NAME,
-            "cors": CORS,
-            "lifecycle": {"rule": LIFECYCLE_RULES},
-            "location": LOCATION,
-            "storageClass": STORAGE_CLASS,
-            "versioning": {"enabled": True},
-            "billing": {"requesterPays": True},
-            "labels": LABELS,
-        }
-
-        connection = _make_connection(DATA)
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-
-        bucket = self._make_one(client=client, name=BUCKET_NAME)
-        bucket.cors = CORS
-        bucket.lifecycle_rules = LIFECYCLE_RULES
-        bucket.storage_class = STORAGE_CLASS
-        bucket.versioning_enabled = True
-        bucket.requester_pays = True
-        bucket.labels = LABELS
-        bucket.create(location=LOCATION)
-
-        connection.api_request.assert_called_once_with(
-            method="POST",
-            path="/b",
-            query_params={"project": PROJECT},
-            data=DATA,
-            _target_object=bucket,
-            timeout=self._get_default_timeout(),
-        )
-
-    def test_create_w_predefined_acl_invalid(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        DATA = {"name": BUCKET_NAME}
-        connection = _Connection(DATA)
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-        bucket = self._make_one(client=client, name=BUCKET_NAME)
-
-        with self.assertRaises(ValueError):
-            bucket.create(predefined_acl="bogus")
-
-    def test_create_w_predefined_acl_valid(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        DATA = {"name": BUCKET_NAME}
-        connection = _Connection(DATA)
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-        bucket = self._make_one(client=client, name=BUCKET_NAME)
-        bucket.create(predefined_acl="publicRead")
-
-        kw, = connection._requested
-        self.assertEqual(kw["method"], "POST")
-        self.assertEqual(kw["path"], "/b")
-        expected_qp = {"project": PROJECT, "predefinedAcl": "publicRead"}
-        self.assertEqual(kw["query_params"], expected_qp)
-        self.assertEqual(kw["data"], DATA)
-        self.assertEqual(kw["timeout"], self._get_default_timeout())
-
-    def test_create_w_predefined_default_object_acl_invalid(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        DATA = {"name": BUCKET_NAME}
-        connection = _Connection(DATA)
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-        bucket = self._make_one(client=client, name=BUCKET_NAME)
-
-        with self.assertRaises(ValueError):
-            bucket.create(predefined_default_object_acl="bogus")
-
-    def test_create_w_predefined_default_object_acl_valid(self):
-        from google.cloud.storage.client import Client
-
-        PROJECT = "PROJECT"
-        BUCKET_NAME = "bucket-name"
-        DATA = {"name": BUCKET_NAME}
-        connection = _Connection(DATA)
-        client = Client(project=PROJECT)
-        client._base_connection = connection
-        bucket = self._make_one(client=client, name=BUCKET_NAME)
-        bucket.create(predefined_default_object_acl="publicRead")
-
-        kw, = connection._requested
-        self.assertEqual(kw["method"], "POST")
-        self.assertEqual(kw["path"], "/b")
-        expected_qp = {"project": PROJECT, "predefinedDefaultObjectAcl": "publicRead"}
-        self.assertEqual(kw["query_params"], expected_qp)
-        self.assertEqual(kw["data"], DATA)
-        self.assertEqual(kw["timeout"], self._get_default_timeout())
-
     def test_acl_property(self):
         from google.cloud.storage.acl import BucketACL
 
@@ -2013,6 +1796,51 @@ def test_versioning_enabled_getter(self):
         bucket = self._make_one(name=NAME, properties=before)
         self.assertEqual(bucket.versioning_enabled, True)
 
+    @mock.patch("warnings.warn")
+    def test_create_deprecated(self, mock_warn):
+        from google.cloud.storage.client import Client
+
+        PROJECT = "PROJECT"
+        BUCKET_NAME = "bucket-name"
+        DATA = {"name": BUCKET_NAME}
+        connection = _make_connection(DATA)
+        client = Client(project=PROJECT)
+        client._base_connection = connection
+
+        bucket = self._make_one(client=client, name=BUCKET_NAME)
+        bucket.create()
+
+        connection.api_request.assert_called_once_with(
+            method="POST",
+            path="/b",
+            query_params={"project": PROJECT},
+            data=DATA,
+            _target_object=bucket,
+            timeout=self._get_default_timeout(),
+        )
+
+        mock_warn.assert_called_with(
+            "Bucket.create() is deprecated and will be removed in future."
+            "Use Client.create_bucket() instead.",
+            PendingDeprecationWarning,
+            stacklevel=1,
+        )
+
+    def test_create_w_user_project(self):
+        from google.cloud.storage.client import Client
+
+        PROJECT = "PROJECT"
+        BUCKET_NAME = "bucket-name"
+        DATA = {"name": BUCKET_NAME}
+        connection = _make_connection(DATA)
+        client = Client(project=PROJECT)
+        client._base_connection = connection
+
+        bucket = self._make_one(client=client, name=BUCKET_NAME)
+        bucket._user_project = "USER_PROJECT"
+        with self.assertRaises(ValueError):
+            bucket.create()
+
     def test_versioning_enabled_setter(self):
         NAME = "name"
         bucket = self._make_one(name=NAME)
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index b3e5874ef..4ba98f82e 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -579,6 +579,44 @@ def test_create_bucket_w_conflict(self):
             timeout=self._get_default_timeout(),
         )
 
+    @mock.patch("warnings.warn")
+    def test_create_requester_pays_deprecated(self, mock_warn):
+        from google.cloud.storage.bucket import Bucket
+
+        project = "PROJECT"
+        credentials = _make_credentials()
+        client = self._make_one(project=project, credentials=credentials)
+        bucket_name = "bucket-name"
+        json_expected = {"name": bucket_name, "billing": {"requesterPays": True}}
+        http = _make_requests_session([_make_json_response(json_expected)])
+        client._http_internal = http
+
+        URI = "/".join(
+            [
+                client._connection.API_BASE_URL,
+                "storage",
+                client._connection.API_VERSION,
+                "b?project=%s" % (project,),
+            ]
+        )
+
+        bucket = client.create_bucket(bucket_name, requester_pays=True)
+
+        self.assertIsInstance(bucket, Bucket)
+        self.assertEqual(bucket.name, bucket_name)
+        self.assertTrue(bucket.requester_pays)
+        http.request.assert_called_once_with(
+            method="POST", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
+        )
+        json_sent = http.request.call_args_list[0][1]["data"]
+        self.assertEqual(json_expected, json.loads(json_sent))
+
+        mock_warn.assert_called_with(
+            "requester_pays arg is deprecated. Use Bucket().requester_pays instead.",
+            PendingDeprecationWarning,
+            stacklevel=1,
+        )
+
     def test_create_bucket_w_predefined_acl_invalid(self):
         project = "PROJECT"
         bucket_name = "bucket-name"
@@ -671,6 +709,100 @@ def test_create_bucket_w_explicit_location(self):
         )
         self.assertEqual(bucket.location, location)
 
+    def test_create_bucket_w_explicit_project(self):
+        from google.cloud.storage.client import Client
+
+        PROJECT = "PROJECT"
+        OTHER_PROJECT = "other-project-123"
+        BUCKET_NAME = "bucket-name"
+        DATA = {"name": BUCKET_NAME}
+        connection = _make_connection(DATA)
+
+        client = Client(project=PROJECT)
+        client._base_connection = connection
+
+        bucket = client.create_bucket(BUCKET_NAME, project=OTHER_PROJECT)
+        connection.api_request.assert_called_once_with(
+            method="POST",
+            path="/b",
+            query_params={"project": OTHER_PROJECT},
+            data=DATA,
+            _target_object=bucket,
+            timeout=self._get_default_timeout(),
+        )
+
+    def test_create_w_extra_properties(self):
+        from google.cloud.storage.client import Client
+        from google.cloud.storage.bucket import Bucket
+
+        BUCKET_NAME = "bucket-name"
+        PROJECT = "PROJECT"
+        CORS = [
+            {
+                "maxAgeSeconds": 60,
+                "methods": ["*"],
+                "origin": ["https://example.com/frontend"],
+                "responseHeader": ["X-Custom-Header"],
+            }
+        ]
+        LIFECYCLE_RULES = [{"action": {"type": "Delete"}, "condition": {"age": 365}}]
+        LOCATION = "eu"
+        LABELS = {"color": "red", "flavor": "cherry"}
+        STORAGE_CLASS = "NEARLINE"
+        DATA = {
+            "name": BUCKET_NAME,
+            "cors": CORS,
+            "lifecycle": {"rule": LIFECYCLE_RULES},
+            "location": LOCATION,
+            "storageClass": STORAGE_CLASS,
+            "versioning": {"enabled": True},
+            "billing": {"requesterPays": True},
+            "labels": LABELS,
+        }
+
+        connection = _make_connection(DATA)
+        client = Client(project=PROJECT)
+        client._base_connection = connection
+
+        bucket = Bucket(client=client, name=BUCKET_NAME)
+        bucket.cors = CORS
+        bucket.lifecycle_rules = LIFECYCLE_RULES
+        bucket.storage_class = STORAGE_CLASS
+        bucket.versioning_enabled = True
+        bucket.requester_pays = True
+        bucket.labels = LABELS
+        client.create_bucket(bucket, location=LOCATION)
+
+        connection.api_request.assert_called_once_with(
+            method="POST",
+            path="/b",
+            query_params={"project": PROJECT},
+            data=DATA,
+            _target_object=bucket,
+            timeout=self._get_default_timeout(),
+        )
+
+    def test_create_hit(self):
+        from google.cloud.storage.client import Client
+
+        PROJECT = "PROJECT"
+        BUCKET_NAME = "bucket-name"
+        DATA = {"name": BUCKET_NAME}
+        connection = _make_connection(DATA)
+        client = Client(project=PROJECT)
+        client._base_connection = connection
+
+        bucket = client.create_bucket(BUCKET_NAME)
+
+        connection.api_request.assert_called_once_with(
+            method="POST",
+            path="/b",
+            query_params={"project": PROJECT},
+            data=DATA,
+            _target_object=bucket,
+            timeout=self._get_default_timeout(),
+        )
+
     def test_create_bucket_w_string_success(self):
         from google.cloud.storage.bucket import Bucket
 
@@ -687,16 +819,15 @@ def test_create_bucket_w_string_success(self):
                 "b?project=%s" % (project,),
             ]
         )
-        json_expected = {"name": bucket_name, "billing": {"requesterPays": True}}
+        json_expected = {"name": bucket_name}
         data = json_expected
         http = _make_requests_session([_make_json_response(data)])
         client._http_internal = http
 
-        bucket = client.create_bucket(bucket_name, requester_pays=True)
+        bucket = client.create_bucket(bucket_name)
 
         self.assertIsInstance(bucket, Bucket)
         self.assertEqual(bucket.name, bucket_name)
-        self.assertTrue(bucket.requester_pays)
         http.request.assert_called_once_with(
             method="POST", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
         )